How invoking remote MCP servers hosted on AWS AgentCore
Overview
Overview
Recently, I've been exploring AWS AgentCore's new capability to host Model Context Protocol (MCP) servers, and I wanted to share my experience with connecting to these remote servers as a client. The Model Context Protocol is an open standard that enables AI assistants to securely connect with external data sources and tools, and AWS AgentCore provides a managed hosting environment for these servers with built-in authentication and scaling capabilities.
In this article, I'll walk through the process of invoking MCP servers hosted on AWS AgentCore Runtime or proxied via AgentCore Gateway, covering different authentication methods, client implementation patterns, and practical considerations. What struck me most about this approach is how it bridges the gap between local development and enterprise-grade deployment while maintaining the flexibility that makes MCP so powerful.
Understanding AWS AgentCore and MCP
Before diving into the implementation details, let's understand what we're working with. AWS AgentCore is Amazon's managed runtime environment for AI agent applications that supports the Model Context Protocol natively. When you deploy an MCP server to AgentCore Runtime or Gateway, you get:
- Managed Infrastructure: No need to worry about scaling, monitoring, or infrastructure management
- Built-in Authentication: OAuth 2.0 and AWS SigV4 authentication out of the box
- Session Isolation: Each client connection gets its own isolated session
- Serverless Scaling: Automatically scales based on demand
The MCP servers deployed on AgentCore expose their tools and resources through a standardized HTTP interface, making them accessible from any MCP-compatible client regardless of where it's running.
Authentication Methods
One of the first challenges I encountered was understanding the authentication options. AWS AgentCore supports several authentication mechanisms for MCP servers:
1. OAuth 2.0 Authentication
This is the most common approach for production deployments. The OAuth flow involves several modes:
Manual Mode: Interactive browser-based authentication
1class OAuth2Handler:
2 def __init__(self, discovery_url: str, client_id: str, client_secret: str = None):
3 self.discovery_url = discovery_url.rstrip('/')
4 self.client_id = client_id
5 self.client_secret = client_secret
6 self.redirect_uri = "http://localhost:3000"
7
8 async def discover_endpoints(self) -> dict:
9 """Discover OAuth 2.0 endpoints using well-known configuration."""
10 async with httpx.AsyncClient() as client:
11 response = await client.get(self.discovery_url, timeout=10.0)
12 if response.status_code == 200:
13 return response.json()
14 raise ValueError(f"Discovery failed: HTTP {response.status_code}")
15
16 async def get_authorization_url(self) -> str:
17 config = await self.discover_endpoints()
18 auth_endpoint = config.get('authorization_endpoint')
19
20 params = {
21 'response_type': 'code',
22 'client_id': self.client_id,
23 'redirect_uri': self.redirect_uri,
24 'scope': 'openid email aws.cognito.signin.user.admin',
25 'state': 'random_state_12345'
26 }
27 return f"{auth_endpoint}?{urlencode(params)}"
28
29 async def exchange_code_for_tokens(self, authorization_code: str) -> dict:
30 config = await self.discover_endpoints()
31 token_endpoint = config.get('token_endpoint')
32
33 data = {
34 'grant_type': 'authorization_code',
35 'client_id': self.client_id,
36 'code': authorization_code,
37 'redirect_uri': self.redirect_uri
38 }
39
40 async with httpx.AsyncClient() as client:
41 response = await client.post(token_endpoint, data=data)
42 if response.status_code == 200:
43 return response.json()
44 raise ValueError(f"Token exchange failed: {response.status_code}")
45
46# Usage example
47async def manual_oauth_flow():
48 handler = OAuth2Handler(
49 discovery_url="https://cognito-idp.us-east-1.amazonaws.com/.../openid-configuration",
50 client_id="your-cognito-client-id"
51 )
52
53 auth_url = await handler.get_authorization_url()
54 webbrowser.open(auth_url)
55
56 # User completes auth and provides callback URL
57 callback_url = input("Enter callback URL: ")
58 code = parse_qs(urlparse(callback_url).query)["code"][0]
59
60 tokens = await handler.exchange_code_for_tokens(code)
61 return tokens['access_token']
Machine-to-Machine Mode: For automated systems using client credentials
1async def get_m2m_token(oauth_handler: OAuth2Handler) -> dict:
2 """Get M2M access token using client_credentials flow."""
3 config = await oauth_handler.discover_endpoints()
4 token_endpoint = config.get('token_endpoint')
5
6 data = {
7 'grant_type': 'client_credentials',
8 'client_id': oauth_handler.client_id,
9 'client_secret': oauth_handler.client_secret,
10 }
11
12 async with httpx.AsyncClient() as client:
13 response = await client.post(token_endpoint, data=data)
14 if response.status_code == 200:
15 return response.json()
16 raise ValueError(f"M2M token request failed: {response.status_code}")
17
18# Usage example
19async def m2m_oauth_flow():
20 handler = OAuth2Handler(
21 discovery_url="https://cognito-idp.us-east-1.amazonaws.com/.../openid-configuration",
22 client_id="your-m2m-client-id",
23 client_secret="your-m2m-client-secret"
24 )
25
26 tokens = await get_m2m_token(handler)
27 return tokens['access_token']
Quick Mode: For AWS Cognito with existing user credentials
1async def cognito_quick_mode(discovery_url: str, client_id: str, username: str, password: str):
2 """Quick token retrieval using AWS Cognito direct authentication."""
3 # Extract region from discovery URL
4 region = re.search(r'cognito-idp\.([^.]+)\.amazonaws\.com', discovery_url).group(1)
5
6 # Use boto3 for direct authentication
7 session = boto3.Session()
8 cognito_client = session.client('cognito-idp', region_name=region)
9
10 response = cognito_client.initiate_auth(
11 ClientId=client_id,
12 AuthFlow='USER_PASSWORD_AUTH',
13 AuthParameters={'USERNAME': username, 'PASSWORD': password}
14 )
15
16 return response['AuthenticationResult']['AccessToken']
2. AWS SigV4 Authentication
For AWS-native integrations, you can use SigV4 signing with your AWS credentials:
1from botocore.auth import SigV4Auth
2from botocore.awsrequest import AWSRequest
3
4class HTTPXSigV4Auth(httpx.Auth):
5 def __init__(self, credentials, service: str, region: str):
6 self.credentials = credentials
7 self.service = service
8 self.region = region
9
10 def auth_flow(self, request: httpx.Request):
11 # Extract request body for signing
12 body = request.content if hasattr(request, 'content') else b''
13
14 # Create AWS request for signing
15 aws_request = AWSRequest(method=request.method, url=str(request.url), data=body)
16 aws_request.headers['Host'] = request.url.host
17
18 # Sign the request
19 signer = SigV4Auth(self.credentials, self.service, self.region)
20 signer.add_auth(aws_request)
21
22 # Update HTTPX request with signed headers
23 for name, value in aws_request.headers.items():
24 request.headers[name] = value
25
26 yield request
27
28class SigV4AgentCoreMCPClient:
29 def __init__(self, agent_arn: str, region: str = "us-west-2"):
30 self.agent_arn = agent_arn
31 self.region = region
32 self.session = boto3.Session()
33 self.credentials = self.session.get_credentials()
34
35 def get_mcp_url(self) -> str:
36 encoded_arn = self.agent_arn.replace(':', '%3A').replace('/', '%2F')
37 return f"https://bedrock-agentcore.{self.region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
38
39 async def connect(self):
40 mcp_url = self.get_mcp_url()
41 auth = HTTPXSigV4Auth(self.credentials, 'bedrock-agentcore', self.region)
42
43 async with streamablehttp_client(url=mcp_url, auth=auth) as (read, write, _):
44 async with ClientSession(read, write) as session:
45 await session.initialize()
46 return session
47
48# Usage example
49async def sigv4_connection_example():
50 client = SigV4AgentCoreMCPClient(
51 agent_arn="arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my-server"
52 )
53 session = await client.connect()
54 return session
Client Implementation Patterns
Based on my experience, here are the key patterns I've found effective for implementing MCP clients that connect to AgentCore-hosted servers:
Basic Client Structure
1async def connect_to_agentcore_server(agent_arn, bearer_token):
2 # Encode the ARN for URL usage
3 encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
4 mcp_url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
5
6 headers = {"authorization": f"Bearer {bearer_token}"}
7
8 async with streamablehttp_client(mcp_url, headers) as (read, write, _):
9 async with ClientSession(read, write) as session:
10 await session.initialize()
11
12 # Discover capabilities
13 tools = await session.list_tools()
14 resources = await session.list_resources()
15
16 return session
17
18# Usage
19async def main():
20 session = await connect_to_agentcore_server(agent_arn, bearer_token)
21 result = await session.call_tool("add_numbers", {"a": 5, "b": 3})
22 print(f"Result: {result}")
MCP Connection Testing
Here's a simplified approach to test MCP connections:
1async def test_mcp_connection(mcp_server_url: str, access_token: str):
2 """Test MCP connection with access token."""
3 headers = {"Authorization": f"Bearer {access_token}"}
4
5 async with streamablehttp_client(mcp_server_url, headers) as (read, write, _):
6 async with ClientSession(read, write) as session:
7 await session.initialize()
8
9 # List available tools and resources
10 tools = await session.list_tools()
11 resources = await session.list_resources()
12
13 print(f"Found {len(tools.tools)} tools and {len(resources.resources)} resources")
14 return session
15
16def load_config() -> dict:
17 """Load configuration from environment variables."""
18 return {
19 'discovery_url': os.getenv('OAUTH_DISCOVERY_URL'),
20 'client_id': os.getenv('OAUTH_CLIENT_ID'),
21 'agentcore_runtime_arn': os.getenv('AGENTCORE_RUNTIME_ARN'),
22 'agentcore_region': os.getenv('AGENTCORE_REGION', 'us-west-2'),
23 }
24
25async def test_oauth_flow(config: dict):
26 """Test OAuth flow and MCP connection."""
27 # Get OAuth token
28 handler = OAuth2Handler(config['discovery_url'], config['client_id'])
29 auth_url = await handler.get_authorization_url()
30 webbrowser.open(auth_url)
31
32 # User provides callback URL
33 callback_url = input("Enter callback URL: ")
34 code = parse_qs(urlparse(callback_url).query)["code"][0]
35
36 tokens = await handler.exchange_code_for_tokens(code)
37
38 # Test MCP connection
39 encoded_arn = config['agentcore_runtime_arn'].replace(':', '%3A').replace('/', '%2F')
40 mcp_url = f"https://bedrock-agentcore.{config['agentcore_region']}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
41
42 session = await test_mcp_connection(mcp_url, tokens['access_token'])
43 return session
Practical Usage Patterns
Tool Discovery and Invocation
MCP enables dynamic discovery of server capabilities:
1async def explore_server_capabilities(session):
2 # Discover available tools and resources
3 tools_response = await session.list_tools()
4 resources_response = await session.list_resources()
5
6 for tool in tools_response.tools:
7 print(f"Tool: {tool.name} - {tool.description}")
8
9 for resource in resources_response.resources:
10 print(f"Resource: {resource.name} ({resource.mimeType})")
11
12async def call_tool_dynamically(session, tool_name, **kwargs):
13 result = await session.call_tool(tool_name, kwargs)
14 return result.content
Resource Access
Access server resources with simple calls:
1async def read_server_resource(session, resource_uri):
2 result = await session.read_resource(resource_uri)
3 return result.contents
4
5# Example usage
6contents = await read_server_resource(session, "file://config.json")
7for content in contents:
8 print(f"{content.mimeType}: {content.text}")
Complete Working Example
Here's a simplified production-ready example:
1async def main():
2 """Main function demonstrating MCP client connection."""
3 config = load_config()
4
5 print("Select authentication mode:")
6 print("1. OAuth 2.0 (Manual)")
7 print("2. AWS Cognito (Quick)")
8 print("3. AWS SigV4")
9
10 choice = input("Choose (1/2/3): ")
11
12 if choice == "1":
13 session = await test_oauth_flow(config)
14 elif choice == "2":
15 token = await cognito_quick_mode(
16 config['discovery_url'],
17 config['client_id'],
18 config['test_username'],
19 config['test_password']
20 )
21 session = await test_mcp_connection(config['mcp_server_url'], token)
22 elif choice == "3":
23 client = SigV4AgentCoreMCPClient(config['agentcore_runtime_arn'])
24 session = await client.connect()
25
26 # Use the session
27 tools = await session.list_tools()
28 print(f"Connected! Found {len(tools.tools)} tools available.")
29
30# Environment setup example
31def setup_environment():
32 """Required environment variables."""
33 env_vars = {
34 'OAUTH_DISCOVERY_URL': 'https://cognito-idp.us-east-1.amazonaws.com/.../openid-configuration',
35 'OAUTH_CLIENT_ID': 'your-cognito-client-id',
36 'AGENTCORE_RUNTIME_ARN': 'arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my-server',
37 'OAUTH_TEST_USERNAME': 'testuser@example.com',
38 'OAUTH_TEST_PASSWORD': 'your-password'
39 }
40
41 for key, example in env_vars.items():
42 print(f"export {key}='{example}'")
43
44if __name__ == "__main__":
45 asyncio.run(main())
Key Learnings
1. Authentication Complexity
The biggest lesson from working with MCP servers on AgentCore is that authentication setup is often the most complex part. Whether you're using OAuth with Cognito, Azure AD, or other providers, getting the client credentials and discovery URLs right is crucial. I recommend starting with the manual OAuth mode for testing before moving to automated flows.
2. Connection Lifecycle Management
Unlike local MCP servers where you might maintain persistent connections, AgentCore-hosted servers require careful attention to connection lifecycle. The platform provides session isolation, but you need to handle reconnection gracefully in your client code.
3. Error Handling is Critical
Remote MCP servers introduce network-related failure modes that don't exist with local servers. Building robust retry logic and graceful degradation from the start saves significant debugging time later.
4. Tool Discovery Enables Dynamic Behavior
One of MCP's most powerful features is tool discovery. Rather than hard-coding tool names and parameters, building clients that dynamically discover and adapt to server capabilities makes your code much more resilient to server updates.
Conclusion
Working with MCP servers hosted on AWS AgentCore opens up exciting possibilities for building distributed AI agent systems. The combination of MCP's flexible protocol with AgentCore's managed infrastructure provides a powerful foundation for enterprise AI applications.
The key to success lies in understanding the authentication flows, building robust connection management, and embracing MCP's dynamic discovery capabilities. While there are complexity challenges, particularly around authentication and error handling, the benefits of managed hosting and automatic scaling make this approach very compelling for production deployments.
As the ecosystem continues to mature, I expect we'll see more standardized client libraries and simplified authentication flows that make this integration even more accessible to developers.