feat: set gateway env vars in agentcore dev for local testing

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

@@ -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/mcp_client/client.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/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]

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

@@ -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/mcp_client/client.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/pyproject.toml

@@ -30,6 +30,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/openaiagents/base/main.py

@@ -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/mcp_client/client.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/pyproject.toml

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

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

@@ -1,7 +1,11 @@
 from strands import Agent, tool
 from bedrock_agentcore.runtime import BedrockAgentCoreApp
 from model.load import load_model
+{{#if hasGateway}}
+from mcp_client.client import get_all_gateway_mcp_clients
+{{else}}
 from mcp_client.client import get_streamable_http_mcp_client
+{{/if}}
 {{#if hasMemory}}
 from memory.session import get_memory_session_manager
 {{/if}}
@@ -10,7 +14,11 @@
 log = app.logger
 
 # Define a Streamable HTTP MCP Client
-mcp_client = get_streamable_http_mcp_client()
+{{#if hasGateway}}
+mcp_clients = get_all_gateway_mcp_clients()
+{{else}}
+mcp_clients = [get_streamable_http_mcp_client()]
+{{/if}}
 
 # Define a collection of tools used by the model
 tools = []
@@ -22,6 +30,11 @@ def add_numbers(a: int, b: int) -> int:
     return a+b
 tools.append(add_numbers)
 
+# Add MCP client to tools if available
+for mcp_client in mcp_clients:
+    if mcp_client:
+        tools.append(mcp_client)
+
 
 {{#if hasMemory}}
 def agent_factory():
@@ -36,7 +49,7 @@ def get_or_create_agent(session_id, user_id):
                 system_prompt="""
                     You are a helpful assistant. Use tools when appropriate.
                 """,
-                tools=tools+[mcp_client]
+                tools=tools
             )
         return cache[key]
     return get_or_create_agent
@@ -52,7 +65,7 @@ def get_or_create_agent():
             system_prompt="""
                 You are a helpful assistant. Use tools when appropriate.
             """,
-            tools=tools+[mcp_client]
+            tools=tools
         )
     return _agent
 {{/if}}