refactor: use mcp-proxy-for-aws instead of custom SigV4 auth

Replace custom SigV4HTTPXAuth class with official mcp-proxy-for-aws:
- Strands: factory pattern with aws_iam_streamablehttp_client
- LangChain: direct session pattern with ClientSession + load_mcp_tools
- OpenAI/ADK/AutoGen: removed custom auth (SDKs don't support custom
  transports, rely on runtime IAM credentials)

Author: aidandaly24

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

@@ -2001,59 +2001,42 @@ exports[`Assets Directory Snapshots > Python framework assets > python/python/la
 exports[`Assets Directory Snapshots > Python framework assets > python/python/langchain_langgraph/base/mcp_client/client.py should match snapshot 1`] = `
 "import os
 import logging
+from mcp.client.streamable_http import streamablehttp_client
 from langchain_mcp_adapters.client import MultiServerMCPClient
 
 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.client.session import ClientSession
+from langchain_mcp_adapters.tools import load_mcp_tools
+from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
 {{/if}}
 
-
-def get_all_gateway_mcp_client() -> MultiServerMCPClient | None:
-    """Returns an MCP Client connected to all configured gateways."""
-    servers = {}
-    {{#each gatewayProviders}}
+{{#each gatewayProviders}}
+{{#if (eq authType "AWS_IAM")}}
+async def get_{{snakeCase name}}_mcp_tools():
+    """Returns MCP tools from the {{name}} gateway using SigV4 auth."""
     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())}
-        {{else}}
-        servers["{{name}}"] = {"transport": "streamable_http", "url": url}
-        {{/if}}
-    else:
+    if not url:
+        logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
+        return []
+    mcp_client = aws_iam_streamablehttp_client(url, aws_service="bedrock-agentcore")
+    async with mcp_client as (read, write, _):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            return await load_mcp_tools(session)
+{{else}}
+def get_{{snakeCase name}}_mcp_client() -> MultiServerMCPClient | None:
+    """Returns an MCP Client for the {{name}} gateway."""
+    url = os.environ.get("{{envVarName}}")
+    if not url:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
-    {{/each}}
-    if not servers:
         return None
-    return MultiServerMCPClient(servers)
+    return MultiServerMCPClient({"{{name}}": {"transport": "streamable_http", "url": url}})
+{{/if}}
+
+{{/each}}
 {{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"
@@ -2454,48 +2437,13 @@ from agents.mcp import MCPServerStreamableHttp
 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
-{{/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())}))
-        {{else}}
         servers.append(MCPServerStreamableHttp(name="{{name}}", params={"url": url}))
-        {{/if}}
     else:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
     {{/each}}
@@ -2786,33 +2734,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 +2745,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}}

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

@@ -1,58 +1,41 @@
 import os
 import logging
+from mcp.client.streamable_http import streamablehttp_client
 from langchain_mcp_adapters.client import MultiServerMCPClient
 
 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.client.session import ClientSession
+from langchain_mcp_adapters.tools import load_mcp_tools
+from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client
 {{/if}}
 
-
-def get_all_gateway_mcp_client() -> MultiServerMCPClient | None:
-    """Returns an MCP Client connected to all configured gateways."""
-    servers = {}
-    {{#each gatewayProviders}}
+{{#each gatewayProviders}}
+{{#if (eq authType "AWS_IAM")}}
+async def get_{{snakeCase name}}_mcp_tools():
+    """Returns MCP tools from the {{name}} gateway using SigV4 auth."""
     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())}
-        {{else}}
-        servers["{{name}}"] = {"transport": "streamable_http", "url": url}
-        {{/if}}
-    else:
+    if not url:
+        logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
+        return []
+    mcp_client = aws_iam_streamablehttp_client(url, aws_service="bedrock-agentcore")
+    async with mcp_client as (read, write, _):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            return await load_mcp_tools(session)
+{{else}}
+def get_{{snakeCase name}}_mcp_client() -> MultiServerMCPClient | None:
+    """Returns an MCP Client for the {{name}} gateway."""
+    url = os.environ.get("{{envVarName}}")
+    if not url:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
-    {{/each}}
-    if not servers:
         return None
-    return MultiServerMCPClient(servers)
+    return MultiServerMCPClient({"{{name}}": {"transport": "streamable_http", "url": url}})
+{{/if}}
+
+{{/each}}
 {{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"

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

@@ -5,48 +5,13 @@
 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
-{{/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())}))
-        {{else}}
         servers.append(MCPServerStreamableHttp(name="{{name}}", params={"url": url}))
-        {{/if}}
     else:
         logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
     {{/each}}

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

@@ -7,33 +7,7 @@
 
 {{#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}}
@@ -44,8 +18,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}}