#!/usr/bin/env python3
"""
Test client for deployed Cost Explorer MCP Server on Agentcore Runtime.
"""
import asyncio
import os
import sys
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def test_remote_server():
"""Test the deployed MCP server 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 or not bearer_token:
print("❌ Error: AGENT_ARN or BEARER_TOKEN environment variable is not set")
print("\n📝 To test your deployed server:")
print("1. Set AGENT_ARN environment variable (or ensure agent_arn.txt exists)")
print("2. Set BEARER_TOKEN environment variable with your OAuth token")
print("\nExample:")
print("export AGENT_ARN='arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/cost-explorer-xyz123'")
print("export BEARER_TOKEN='your-oauth-token'")
sys.exit(1)
# Encode ARN for URL
encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
mcp_url = f"https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
headers = {
"authorization": f"Bearer {bearer_token}",
"Content-Type": "application/json"
}
print(f"🔗 Testing deployed MCP server...")
print(f"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
await session.initialize()
print("✅ Remote session initialized successfully")
# List 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}")
# 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 remote tests passed! Deployed MCP server is working correctly.")
except Exception as e:
print(f"❌ Error testing remote MCP server: {e}")
print("\n🔍 Troubleshooting tips:")
print("1. Verify your BEARER_TOKEN is valid and not expired")
print("2. Check that the AGENT_ARN is correct")
print("3. Ensure your OAuth configuration is properly set up")
print("4. Verify the server is deployed and running")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(test_remote_server())
