#!/usr/bin/env python3
"""
MCP Client for testing the deployed Cost Explorer MCP Server on AgentCore Runtime.
Based on AWS documentation: https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-mcp.html
"""
import asyncio
import os
import sys
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def test_agentcore_mcp_server():
"""Test the MCP server deployed on AgentCore Runtime."""
# Get environment variables
agent_arn = os.getenv('AGENT_ARN')
bearer_token = os.getenv('BEARER_TOKEN')
# Try to read ARN from file if not in environment
if not agent_arn and os.path.exists('agent_arn.txt'):
with open('agent_arn.txt', 'r') as f:
agent_arn = f.read().strip()
if not agent_arn:
print("❌ Error: AGENT_ARN environment variable is not set and agent_arn.txt not found")
print("\n📝 To test your deployed MCP server:")
print("1. Set AGENT_ARN environment variable:")
print(" export AGENT_ARN='arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/cost-explorer-xyz123'")
print("2. Set BEARER_TOKEN environment variable with your OAuth token (if using OAuth):")
print(" export BEARER_TOKEN='your-oauth-token'")
print("3. Or configure Cognito authentication as described in AWS docs")
sys.exit(1)
# Encode ARN for URL (replace : with %3A and / with %2F)
encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
# Construct the MCP endpoint URL for AgentCore Runtime
# Based on AWS docs: https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT
region = "us-west-2" # Extract from ARN if needed
mcp_url = f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
# Set up headers
headers = {"Content-Type": "application/json"}
# Add Bearer token if provided
if bearer_token:
headers["authorization"] = f"Bearer {bearer_token}"
print(f"🔐 Using Bearer token authentication")
else:
print("⚠️ No Bearer token provided - using default authentication")
print(" For production use, you'll need OAuth authentication tokens")
print(f"🔗 Testing MCP server on AgentCore Runtime...")
print(f"Agent ARN: {agent_arn}")
print(f"MCP URL: {mcp_url}")
print(f"Headers: {headers}")
try:
async with streamablehttp_client(
mcp_url,
headers,
timeout=120,
terminate_on_close=False
) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
# Initialize the session
print("🔄 Initializing MCP session...")
await session.initialize()
print("✅ MCP session initialized successfully")
# List available tools
print("🔄 Listing available tools...")
tool_result = await session.list_tools()
print(f"✅ Available tools: {len(tool_result.tools)}")
for tool in tool_result.tools:
print(f" - {tool.name}: {tool.description[:100]}...")
# Test a simple tool call
print("\n🧪 Testing get_today_date tool...")
result = await session.call_tool("get_today_date", {})
print(f"✅ get_today_date result: {result.content[0].text}")
print("\n✅ All MCP tests passed! AgentCore Runtime MCP server is working correctly.")
except Exception as e:
print(f"❌ Error testing AgentCore Runtime MCP server: {e}")
print("\n🔍 Troubleshooting tips:")
print("1. Verify the MCP server is deployed and running on AgentCore Runtime")
print("2. Check that the agent ARN is correct")
print("3. Ensure you have proper AWS credentials and permissions")
print("4. Set up OAuth authentication as described in AWS documentation:")
print(" https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/runtime-mcp.html")
print("5. The MCP server should be accessible at port 8000 within the runtime")
print("6. Try setting up Cognito user pool for authentication")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(test_agentcore_mcp_server())
