refactor: use mcp-proxy-for-aws for gateway auth, remove AutoGen gateway support

Replace custom SigV4HTTPXAuth class with official mcp-proxy-for-aws package:
- Strands: aws_iam_streamablehttp_client factory pattern
- LangChain: SigV4HTTPXAuth via auth param in MultiServerMCPClient config
- OpenAI Agents: SigV4HTTPXAuth via httpx_client_factory param
- Google ADK: SigV4HTTPXAuth via httpx_client_factory in StreamableHTTPConnectionParams

Revert AutoGen to original upstream — SDK doesn't support custom
httpx auth (no httpx_client_factory param).

Author: aidandaly24

src/assets/__tests__/__snapshots__/assets.snapshot.test.ts.snap

@@ -918,11 +918,7 @@ from autogen_agentchat.agents import AssistantAgent
 from autogen_core.tools import FunctionTool
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
-{{#if hasGateway}}
-from mcp_client.client import get_all_gateway_mcp_tools
-{{else}}
 from mcp_client.client import get_streamable_http_mcp_tools
-{{/if}}
 
 app = BedrockAgentCoreApp()
 log = app.logger
@@ -947,13 +943,7 @@ async def invoke(payload, context):
     log.info("Invoking Agent.....")
 
     # Get MCP Tools
-    {{#if hasGateway}}
-    mcp_tools = await get_all_gateway_mcp_tools()
-    {{else}}
     mcp_tools = await get_streamable_http_mcp_tools()
-    {{/if}}
-    if mcp_tools is None:
-        mcp_tools = []
 
     # Define an AssistantAgent with the model and tools
     agent = AssistantAgent(
@@ -984,41 +974,24 @@ exports[`Assets Directory Snapshots > Python framework assets > python/python/au
 `;
 
 exports[`Assets Directory Snapshots > Python framework assets > python/python/autogen/base/mcp_client/client.py should match snapshot 1`] = `
-"import os
-import logging
-from typing import List
+"from typing import List
 from autogen_ext.tools.mcp import (
     StreamableHttpMcpToolAdapter,
     StreamableHttpServerParams,
     mcp_server_tools,
 )
 
-logger = logging.getLogger(__name__)
-
-{{#if hasGateway}}
-async def get_all_gateway_mcp_tools() -> List[StreamableHttpMcpToolAdapter]:
-    """Returns MCP Tools from all configured gateways."""
-    tools = []
-    {{#each gatewayProviders}}
-    url = os.environ.get("{{envVarName}}")
-    if url:
-        server_params = StreamableHttpServerParams(url=url)
-        tools.extend(await mcp_server_tools(server_params))
-    else:
-        logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
-    {{/each}}
-    return tools
-{{else}}
 # ExaAI provides information about code through web searches, crawling and code context searches through their platform. Requires no authentication
 EXAMPLE_MCP_ENDPOINT = "https://mcp.exa.ai/mcp"
 
 
 async def get_streamable_http_mcp_tools() -> List[StreamableHttpMcpToolAdapter]:
-    """Returns MCP Tools compatible with AutoGen."""
+    """
+    Returns MCP Tools compatible with AutoGen.
+    """
     # to use an MCP server that supports bearer authentication, add headers={"Authorization": f"Bearer {access_token}"}
     server_params = StreamableHttpServerParams(url=EXAMPLE_MCP_ENDPOINT)
     return await mcp_server_tools(server_params)
-{{/if}}
 "
 `;
 
@@ -1735,13 +1708,27 @@ from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnecti
 logger = logging.getLogger(__name__)
 
 {{#if hasGateway}}
+{{#if (includes gatewayAuthTypes "AWS_IAM")}}
+import httpx
+from mcp_proxy_for_aws.sigv4_helper import SigV4HTTPXAuth, create_aws_session
+{{/if}}
+
 def get_all_gateway_mcp_toolsets() -> list[MCPToolset]:
     """Returns MCP Toolsets for all configured gateways."""
     toolsets = []
     {{#each gatewayProviders}}
     url = os.environ.get("{{envVarName}}")
     if url:
+        {{#if (eq authType "AWS_IAM")}}
+        session = create_aws_session()
+        auth = SigV4HTTPXAuth(session.get_credentials(), "bedrock-agentcore", session.region_name)
+        toolsets.append(MCPToolset(connection_params=StreamableHTTPConnectionParams(
+            url=url,
+            httpx_client_factory=lambda **kwargs: httpx.AsyncClient(auth=auth, **kwargs)
+        )))
+        {{else}}
         toolsets.append(MCPToolset(connection_params=StreamableHTTPConnectionParams(url=url)))
+        {{/if}}
     else:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
     {{/each}}
@@ -1828,6 +1815,8 @@ dependencies = [
     "google-adk >= 1.17.0",
     "bedrock-agentcore >= 1.0.3",
     "botocore[crt] >= 1.35.0",
+    {{#if hasGateway}}{{#if (includes gatewayAuthTypes "AWS_IAM")}}"mcp-proxy-for-aws >= 1.1.0",
+    {{/if}}{{/if}}
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -2007,44 +1996,19 @@ logger = logging.getLogger(__name__)
 
 {{#if hasGateway}}
 {{#if (includes gatewayAuthTypes "AWS_IAM")}}
-import boto3
-import httpx
-from botocore.auth import SigV4Auth
-from botocore.awsrequest import AWSRequest
-
-
-class SigV4HTTPXAuth(httpx.Auth):
-    """Signs HTTP requests with AWS SigV4 for Lambda function URL authentication."""
-
-    def __init__(self):
-        session = boto3.Session()
-        credentials = session.get_credentials().get_frozen_credentials()
-        region = session.region_name or os.environ.get("AWS_REGION", "us-east-1")
-        self.signer = SigV4Auth(credentials, "lambda", region)
-
-    def auth_flow(self, request):
-        headers = dict(request.headers)
-        headers.pop("connection", None)
-        aws_request = AWSRequest(
-            method=request.method,
-            url=str(request.url),
-            data=request.content,
-            headers=headers,
-        )
-        self.signer.add_auth(aws_request)
-        request.headers.update(dict(aws_request.headers))
-        yield request
+from mcp_proxy_for_aws.sigv4_helper import SigV4HTTPXAuth, create_aws_session
 {{/if}}
 
-
 def get_all_gateway_mcp_client() -> MultiServerMCPClient | None:
     """Returns an MCP Client connected to all configured gateways."""
     servers = {}
     {{#each gatewayProviders}}
     url = os.environ.get("{{envVarName}}")
     if url:
         {{#if (eq authType "AWS_IAM")}}
-        servers["{{name}}"] = {"transport": "streamable_http", "url": url, "http_client": httpx.AsyncClient(auth=SigV4HTTPXAuth())}
+        session = create_aws_session()
+        auth = SigV4HTTPXAuth(session.get_credentials(), "bedrock-agentcore", session.region_name)
+        servers["{{name}}"] = {"transport": "streamable_http", "url": url, "auth": auth}
         {{else}}
         servers["{{name}}"] = {"transport": "streamable_http", "url": url}
         {{/if}}
@@ -2239,6 +2203,8 @@ dependencies = [
 {{#if (eq modelProvider "Gemini")}}
     "langchain-google-genai >= 3.0.3",
 {{/if}}
+    {{#if hasGateway}}{{#if (includes gatewayAuthTypes "AWS_IAM")}}"mcp-proxy-for-aws >= 1.1.0",
+    {{/if}}{{/if}}
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -2455,44 +2421,23 @@ logger = logging.getLogger(__name__)
 
 {{#if hasGateway}}
 {{#if (includes gatewayAuthTypes "AWS_IAM")}}
-import boto3
 import httpx
-from botocore.auth import SigV4Auth
-from botocore.awsrequest import AWSRequest
-
-
-class SigV4HTTPXAuth(httpx.Auth):
-    """Signs HTTP requests with AWS SigV4 for Lambda function URL authentication."""
-
-    def __init__(self):
-        session = boto3.Session()
-        credentials = session.get_credentials().get_frozen_credentials()
-        region = session.region_name or os.environ.get("AWS_REGION", "us-east-1")
-        self.signer = SigV4Auth(credentials, "lambda", region)
-
-    def auth_flow(self, request):
-        headers = dict(request.headers)
-        headers.pop("connection", None)
-        aws_request = AWSRequest(
-            method=request.method,
-            url=str(request.url),
-            data=request.content,
-            headers=headers,
-        )
-        self.signer.add_auth(aws_request)
-        request.headers.update(dict(aws_request.headers))
-        yield request
+from mcp_proxy_for_aws.sigv4_helper import SigV4HTTPXAuth, create_aws_session
 {{/if}}
 
-
 def get_all_gateway_mcp_servers() -> list[MCPServerStreamableHttp]:
     """Returns MCP servers for all configured gateways."""
     servers = []
     {{#each gatewayProviders}}
     url = os.environ.get("{{envVarName}}")
     if url:
         {{#if (eq authType "AWS_IAM")}}
-        servers.append(MCPServerStreamableHttp(name="{{name}}", params={"url": url, "http_client": httpx.AsyncClient(auth=SigV4HTTPXAuth())}))
+        session = create_aws_session()
+        auth = SigV4HTTPXAuth(session.get_credentials(), "bedrock-agentcore", session.region_name)
+        servers.append(MCPServerStreamableHttp(
+            name="{{name}}",
+            params={"url": url, "httpx_client_factory": lambda **kwargs: httpx.AsyncClient(auth=auth, **kwargs)}
+        ))
         {{else}}
         servers.append(MCPServerStreamableHttp(name="{{name}}", params={"url": url}))
         {{/if}}
@@ -2577,6 +2522,8 @@ dependencies = [
     "openai-agents >= 0.4.2",
     "bedrock-agentcore >= 1.0.3",
     "botocore[crt] >= 1.35.0",
+    {{#if hasGateway}}{{#if (includes gatewayAuthTypes "AWS_IAM")}}"mcp-proxy-for-aws >= 1.1.0",
+    {{/if}}{{/if}}
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -2786,33 +2733,7 @@ logger = logging.getLogger(__name__)
 
 {{#if hasGateway}}
 {{#if (includes gatewayAuthTypes "AWS_IAM")}}
-import boto3
-import httpx
-from botocore.auth import SigV4Auth
-from botocore.awsrequest import AWSRequest
-
-
-class SigV4HTTPXAuth(httpx.Auth):
-    """Signs HTTP requests with AWS SigV4 for Lambda function URL authentication."""
-
-    def __init__(self):
-        session = boto3.Session()
-        credentials = session.get_credentials().get_frozen_credentials()
-        region = session.region_name or os.environ.get("AWS_REGION", "us-east-1")
-        self.signer = SigV4Auth(credentials, "lambda", region)
-
-    def auth_flow(self, request):
-        headers = dict(request.headers)
-        headers.pop("connection", None)
-        aws_request = AWSRequest(
-            method=request.method,
-            url=str(request.url),
-            data=request.content,
-            headers=headers,
-        )
-        self.signer.add_auth(aws_request)
-        request.headers.update(dict(aws_request.headers))
-        yield request
+from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
 {{/if}}
 
 {{#each gatewayProviders}}
@@ -2823,8 +2744,7 @@ def get_{{snakeCase name}}_mcp_client() -> MCPClient | None:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
         return None
     {{#if (eq authType "AWS_IAM")}}
-    http_client = httpx.AsyncClient(auth=SigV4HTTPXAuth())
-    return MCPClient(lambda: streamablehttp_client(url, http_client=http_client))
+    return MCPClient(lambda: aws_iam_streamablehttp_client(url, aws_service="bedrock-agentcore"))
     {{else}}
     return MCPClient(lambda: streamablehttp_client(url))
     {{/if}}
@@ -3003,6 +2923,8 @@ dependencies = [
     {{/if}}"mcp >= 1.19.0",
     {{#if (eq modelProvider "OpenAI")}}"openai >= 1.0.0",
     {{/if}}"strands-agents >= 1.13.0",
+    {{#if hasGateway}}{{#if (includes gatewayAuthTypes "AWS_IAM")}}"mcp-proxy-for-aws >= 1.1.0",
+    {{/if}}{{/if}}
 ]
 
 [tool.hatch.build.targets.wheel]

src/assets/python/autogen/base/main.py

@@ -3,11 +3,7 @@
 from autogen_core.tools import FunctionTool
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
-{{#if hasGateway}}
-from mcp_client.client import get_all_gateway_mcp_tools
-{{else}}
 from mcp_client.client import get_streamable_http_mcp_tools
-{{/if}}
 
 app = BedrockAgentCoreApp()
 log = app.logger
@@ -32,13 +28,7 @@ async def invoke(payload, context):
     log.info("Invoking Agent.....")
 
     # Get MCP Tools
-    {{#if hasGateway}}
-    mcp_tools = await get_all_gateway_mcp_tools()
-    {{else}}
     mcp_tools = await get_streamable_http_mcp_tools()
-    {{/if}}
-    if mcp_tools is None:
-        mcp_tools = []
 
     # Define an AssistantAgent with the model and tools
     agent = AssistantAgent(

src/assets/python/autogen/base/mcp_client/client.py

@@ -1,35 +1,18 @@
-import os
-import logging
 from typing import List
 from autogen_ext.tools.mcp import (
     StreamableHttpMcpToolAdapter,
     StreamableHttpServerParams,
     mcp_server_tools,
 )
 
-logger = logging.getLogger(__name__)
-
-{{#if hasGateway}}
-async def get_all_gateway_mcp_tools() -> List[StreamableHttpMcpToolAdapter]:
-    """Returns MCP Tools from all configured gateways."""
-    tools = []
-    {{#each gatewayProviders}}
-    url = os.environ.get("{{envVarName}}")
-    if url:
-        server_params = StreamableHttpServerParams(url=url)
-        tools.extend(await mcp_server_tools(server_params))
-    else:
-        logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
-    {{/each}}
-    return tools
-{{else}}
 # ExaAI provides information about code through web searches, crawling and code context searches through their platform. Requires no authentication
 EXAMPLE_MCP_ENDPOINT = "https://mcp.exa.ai/mcp"
 
 
 async def get_streamable_http_mcp_tools() -> List[StreamableHttpMcpToolAdapter]:
-    """Returns MCP Tools compatible with AutoGen."""
+    """
+    Returns MCP Tools compatible with AutoGen.
+    """
     # to use an MCP server that supports bearer authentication, add headers={"Authorization": f"Bearer {access_token}"}
     server_params = StreamableHttpServerParams(url=EXAMPLE_MCP_ENDPOINT)
     return await mcp_server_tools(server_params)
-{{/if}}

src/assets/python/googleadk/base/mcp_client/client.py

@@ -6,13 +6,27 @@
 logger = logging.getLogger(__name__)
 
 {{#if hasGateway}}
+{{#if (includes gatewayAuthTypes "AWS_IAM")}}
+import httpx
+from mcp_proxy_for_aws.sigv4_helper import SigV4HTTPXAuth, create_aws_session
+{{/if}}
+
 def get_all_gateway_mcp_toolsets() -> list[MCPToolset]:
     """Returns MCP Toolsets for all configured gateways."""
     toolsets = []
     {{#each gatewayProviders}}
     url = os.environ.get("{{envVarName}}")
     if url:
+        {{#if (eq authType "AWS_IAM")}}
+        session = create_aws_session()
+        auth = SigV4HTTPXAuth(session.get_credentials(), "bedrock-agentcore", session.region_name)
+        toolsets.append(MCPToolset(connection_params=StreamableHTTPConnectionParams(
+            url=url,
+            httpx_client_factory=lambda **kwargs: httpx.AsyncClient(auth=auth, **kwargs)
+        )))
+        {{else}}
         toolsets.append(MCPToolset(connection_params=StreamableHTTPConnectionParams(url=url)))
+        {{/if}}
     else:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
     {{/each}}

src/assets/python/googleadk/base/pyproject.toml

@@ -14,6 +14,8 @@ dependencies = [
     "google-adk >= 1.17.0",
     "bedrock-agentcore >= 1.0.3",
     "botocore[crt] >= 1.35.0",
+    {{#if hasGateway}}{{#if (includes gatewayAuthTypes "AWS_IAM")}}"mcp-proxy-for-aws >= 1.1.0",
+    {{/if}}{{/if}}
 ]
 
 [tool.hatch.build.targets.wheel]