amazon-bedrock-agentcore-samples/02-use-cases/healthcare-appointment-agent/utils.py

import httpx
import os
from dotenv import load_dotenv
import boto3
import requests

#Reading environment variables
load_dotenv()

def create_agentcore_client():
    #create boto3 session and client
    if os.getenv("awscred_profile_name") is None:
        boto_session = boto3.Session(region_name=os.getenv("aws_default_region")) #using default profile
    else:
        boto_session = boto3.Session(profile_name=os.getenv("awscred_profile_name"), region_name=os.getenv("aws_default_region"))

    agentcore_client = boto_session.client(
        "bedrock-agentcore-control"
    )

    return boto_session, agentcore_client

def create_http_client(**kwargs) -> httpx.AsyncClient:
    """Create an HTTPX client with SSL verification disabled."""
    kwargs['verify'] = False
    # Optionally add other configurations
    kwargs.setdefault('timeout', httpx.Timeout(30.0))
    return httpx.AsyncClient(**kwargs)

def get_gateway_endpoint(agentcore_client, gateway_id):
    response = agentcore_client.get_gateway(
        gatewayIdentifier=gateway_id\
    )

    return response['gatewayUrl']

def list_gateways(agentcore_client):
    response = agentcore_client.list_gateways()

    return response['items']

def get_oath_token(boto_session):
    response = requests.post(
        os.getenv("cognito_token_url"),
        data=f"grant_type=client_credentials&client_id={os.getenv('cognito_client_id')}&client_secret={get_cognito_client_secret(boto_session)}&scope={os.getenv('cognito_auth_scope')}",
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )

    #print(response.json())
    return response.json()['access_token']

def get_cognito_client_secret(boto_session):
    cognito_client = boto_session.client("cognito-idp", region_name=os.getenv('aws_default_region'))

    print(f"Getting Client Secret using UserPoolId: {os.getenv('cognito_user_pool_id')} and ClientId: {os.getenv('cognito_client_id')}")
    response = cognito_client.describe_user_pool_client(
        UserPoolId=os.getenv('cognito_user_pool_id'),
        ClientId=os.getenv('cognito_client_id')
    )
    return response['UserPoolClient']['ClientSecret']