35 lines
1.4 KiB
Python
35 lines
1.4 KiB
Python
import os
|
|
import logging
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
from strands.tools.mcp.mcp_client import MCPClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_nmpenterprisetools_mcp_client() -> MCPClient | None:
|
|
"""Returns an MCP Client connected to the nmpEnterpriseTools gateway."""
|
|
url = os.environ.get("AGENTCORE_GATEWAY_NMPENTERPRISETOOLS_URL")
|
|
if not url:
|
|
logger.warning("AGENTCORE_GATEWAY_NMPENTERPRISETOOLS_URL not set — nmpEnterpriseTools gateway tools unavailable")
|
|
return None
|
|
return MCPClient(lambda: streamablehttp_client(url), prefix="nmpenterprisetools")
|
|
|
|
def get_awsinfrastructuretools_mcp_client() -> MCPClient | None:
|
|
"""Returns an MCP Client connected to the awsInfrastructureTools gateway."""
|
|
url = os.environ.get("AGENTCORE_GATEWAY_AWSINFRASTRUCTURETOOLS_URL")
|
|
if not url:
|
|
logger.warning("AGENTCORE_GATEWAY_AWSINFRASTRUCTURETOOLS_URL not set — awsInfrastructureTools gateway tools unavailable")
|
|
return None
|
|
return MCPClient(lambda: streamablehttp_client(url), prefix="awsinfrastructuretools")
|
|
|
|
def get_all_gateway_mcp_clients() -> list[MCPClient]:
|
|
"""Returns MCP clients for all configured gateways."""
|
|
clients = []
|
|
client = get_nmpenterprisetools_mcp_client()
|
|
if client:
|
|
clients.append(client)
|
|
client = get_awsinfrastructuretools_mcp_client()
|
|
if client:
|
|
clients.append(client)
|
|
return clients
|