feat: add gateway auth and multi-gateway support to agent templates (#427)

* feat: add gateway auth support to agent templates

Add SigV4 authentication to MCP client templates so agents can
authenticate with AWS_IAM gateways. Each framework's client.py
uses Handlebars conditionals to include auth when gateways exist.

SigV4HTTPXAuth class signs HTTP requests using botocore SigV4Auth,
passed to the MCP client via httpx.AsyncClient. Templates read
gateway URLs from AGENTCORE_GATEWAY_{NAME}_URL env vars and handle
missing vars gracefully (warn, don't crash).

Updated all 5 frameworks: Strands, LangChain, OpenAI Agents,
Google ADK, AutoGen. Schema mapper reads mcp.json to populate
gateway config for template rendering. All gateways are auto-
included when creating an agent.

* feat: add multi-gateway support and fix template rendering

Replace single-gateway [0] indexing with {{#each gatewayProviders}}
loops. Each gateway gets its own client function (Strands) or entry
in the servers dict (LangChain/OpenAI/AutoGen/ADK).

Add snakeCase Handlebars helper for gateway function names.
Add gatewayAuthTypes array for conditional SigV4 imports.
Fix @index parse error by using plain variable names.

* fix: update main.py files with gateway conditional imports

All 5 framework main.py files now use Handlebars conditionals to
import the correct MCP client function based on hasGateway flag.
Fix snakeCase helper to handle all special characters.

* style: fix formatting

* 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).

* fix: pass AWS region to aws_iam_streamablehttp_client in Strands template

Author: aidandaly24

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

@@ -5,7 +5,11 @@
 from google.genai import types
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
+{{#if hasGateway}}
+from mcp_client.client import get_all_gateway_mcp_toolsets
+{{else}}
 from mcp_client.client import get_streamable_http_mcp_client
+{{/if}}
 
 app = BedrockAgentCoreApp()
 log = app.logger
@@ -23,7 +27,12 @@ def add_numbers(a: int, b: int) -> int:
 
 
 # Get MCP Toolset
-mcp_toolset = [get_streamable_http_mcp_client()]
+{{#if hasGateway}}
+mcp_toolset = get_all_gateway_mcp_toolsets()
+{{else}}
+mcp_client = get_streamable_http_mcp_client()
+mcp_toolset = [mcp_client] if mcp_client else []
+{{/if}}
 
 _credentials_loaded = False
 

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

@@ -1,15 +1,45 @@
+import os
+import logging
 from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset
 from google.adk.tools.mcp_tool.mcp_session_manager import StreamableHTTPConnectionParams
 
+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}}
+    return toolsets
+{{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"
 
 
 def get_streamable_http_mcp_client() -> MCPToolset:
-    """
-    Returns an MCP Toolset compatible with Google ADK.
-    """
+    """Returns an MCP Toolset compatible with Google ADK."""
     # to use an MCP server that supports bearer authentication, add headers={"Authorization": f"Bearer {access_token}"}
     return MCPToolset(
         connection_params=StreamableHTTPConnectionParams(url=EXAMPLE_MCP_ENDPOINT)
     )
+{{/if}}

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

@@ -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]

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

@@ -4,7 +4,11 @@
 from langchain.tools import tool
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
+{{#if hasGateway}}
+from mcp_client.client import get_all_gateway_mcp_client
+{{else}}
 from mcp_client.client import get_streamable_http_mcp_client
+{{/if}}
 
 app = BedrockAgentCoreApp()
 log = app.logger
@@ -34,10 +38,16 @@ async def invoke(payload, context):
     log.info("Invoking Agent.....")
 
     # Get MCP Client
+    {{#if hasGateway}}
+    mcp_client = get_all_gateway_mcp_client()
+    {{else}}
     mcp_client = get_streamable_http_mcp_client()
+    {{/if}}
 
     # Load MCP Tools
-    mcp_tools = await mcp_client.get_tools()
+    mcp_tools = []
+    if mcp_client:
+        mcp_tools = await mcp_client.get_tools()
 
     # Define the agent using create_react_agent
     graph = create_react_agent(get_or_create_model(), tools=mcp_tools + tools)

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

@@ -1,13 +1,40 @@
+import os
+import logging
 from langchain_mcp_adapters.client import MultiServerMCPClient
 
+logger = logging.getLogger(__name__)
+
+{{#if hasGateway}}
+{{#if (includes gatewayAuthTypes "AWS_IAM")}}
+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")}}
+        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}}
+    else:
+        logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
+    {{/each}}
+    if not servers:
+        return None
+    return MultiServerMCPClient(servers)
+{{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"
 
 
 def get_streamable_http_mcp_client() -> MultiServerMCPClient:
-    """
-    Returns an MCP Client compatible with LangChain/LangGraph.
-    """
+    """Returns an MCP Client compatible with LangChain/LangGraph."""
     # to use an MCP server that supports bearer authentication, add headers={"Authorization": f"Bearer {access_token}"}
     return MultiServerMCPClient(
         {
@@ -17,3 +44,4 @@ def get_streamable_http_mcp_client() -> MultiServerMCPClient:
             }
         }
     )
+{{/if}}

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

@@ -29,6 +29,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]

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

@@ -2,13 +2,22 @@
 from agents import Agent, Runner, function_tool
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
+{{#if hasGateway}}
+from mcp_client.client import get_all_gateway_mcp_servers
+{{else}}
 from mcp_client.client import get_streamable_http_mcp_client
+{{/if}}
 
 app = BedrockAgentCoreApp()
 log = app.logger
 
 # Get MCP Server
+{{#if hasGateway}}
+mcp_servers = get_all_gateway_mcp_servers()
+{{else}}
 mcp_server = get_streamable_http_mcp_client()
+mcp_servers = [mcp_server] if mcp_server else []
+{{/if}}
 
 _credentials_loaded = False
 
@@ -30,16 +39,47 @@ def add_numbers(a: int, b: int) -> int:
 async def main(query):
     ensure_credentials_loaded()
     try:
-        async with mcp_server as server:
-            active_servers = [server] if server else []
+        {{#if hasGateway}}
+        if mcp_servers:
             agent = Agent(
                 name="{{ name }}",
                 model="gpt-4.1",
-                mcp_servers=active_servers,
+                mcp_servers=mcp_servers,
                 tools=[add_numbers]
             )
             result = await Runner.run(agent, query)
             return result
+        else:
+            agent = Agent(
+                name="{{ name }}",
+                model="gpt-4.1",
+                mcp_servers=[],
+                tools=[add_numbers]
+            )
+            result = await Runner.run(agent, query)
+            return result
+        {{else}}
+        if mcp_servers:
+            async with mcp_servers[0] as server:
+                active_servers = [server]
+                agent = Agent(
+                    name="{{ name }}",
+                    model="gpt-4.1",
+                    mcp_servers=active_servers,
+                    tools=[add_numbers]
+                )
+                result = await Runner.run(agent, query)
+                return result
+        else:
+            agent = Agent(
+                name="{{ name }}",
+                model="gpt-4.1",
+                mcp_servers=[],
+                tools=[add_numbers]
+            )
+            result = await Runner.run(agent, query)
+            return result
+        {{/if}}
     except Exception as e:
         log.error(f"Error during agent execution: {e}", exc_info=True)
         raise e

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

@@ -1,14 +1,44 @@
+import os
+import logging
 from agents.mcp import MCPServerStreamableHttp
 
+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_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")}}
+        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}}
+    else:
+        logger.warning("{{envVarName}} not set — {{name}} gateway tools unavailable")
+    {{/each}}
+    return servers
+{{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"
 
 
 def get_streamable_http_mcp_client() -> MCPServerStreamableHttp:
-    """
-    Returns an MCP Client compatible with OpenAI Agents SDK.
-    """
+    """Returns an MCP Client compatible with OpenAI Agents SDK."""
     # to use an MCP server that supports bearer authentication, add headers={"Authorization": f"Bearer {access_token}"}
     return MCPServerStreamableHttp(
         name="AgentCore Gateway MCP", params={"url": EXAMPLE_MCP_ENDPOINT}
     )
+{{/if}}

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

@@ -13,6 +13,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]