#!/usr/bin/env python3
"""
Detailed OAuth test for AgentCore Runtime MCP server.
"""
import asyncio
import httpx
import json
import os
import traceback
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def test_http_first():
"""Test HTTP response first to see what we get."""
agent_arn = None
if os.path.exists('agent_arn.txt'):
with open('agent_arn.txt', 'r') as f:
agent_arn = f.read().strip()
if not agent_arn:
print("❌ No agent ARN found")
return
# Get fresh token
bearer_token = os.getenv('BEARER_TOKEN')
if not bearer_token:
print("❌ No BEARER_TOKEN environment variable")
return
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 = {
"Content-Type": "application/json",
"Authorization": f"Bearer {bearer_token}",
"Accept": "application/json"
}
print(f"🔗 Testing HTTP endpoint: {mcp_url}")
print(f"🔐 Using Bearer token: {bearer_token[:50]}...")
# Test with a simple MCP initialize message
mcp_init = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(mcp_url, headers=headers, json=mcp_init, timeout=30.0)
print(f"✅ HTTP Status: {response.status_code}")
print(f"📋 Response Headers: {dict(response.headers)}")
if response.text:
try:
json_response = response.json()
print(f"📄 JSON Response: {json.dumps(json_response, indent=2)}")
except:
print(f"📄 Raw Response: {response.text}")
else:
print("📄 Empty response body")
except Exception as e:
print(f"❌ HTTP Error: {e}")
traceback.print_exc()
async def test_mcp_client():
"""Test with MCP client."""
agent_arn = None
if os.path.exists('agent_arn.txt'):
with open('agent_arn.txt', 'r') as f:
agent_arn = f.read().strip()
bearer_token = os.getenv('BEARER_TOKEN')
if not agent_arn or not bearer_token:
print("❌ Missing agent ARN or bearer token")
return
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 = {
"Content-Type": "application/json",
"Authorization": f"Bearer {bearer_token}",
"Accept": "application/json"
}
print(f"\n🔗 Testing MCP client: {mcp_url}")
try:
async with streamablehttp_client(
mcp_url,
headers,
timeout=30,
terminate_on_close=False
) as (read_stream, write_stream, _):
async with ClientSession(read_stream, write_stream) as session:
print("🔄 Initializing MCP session...")
await session.initialize()
print("✅ MCP session initialized!")
# List tools
tools = await session.list_tools()
print(f"✅ Found {len(tools.tools)} tools:")
for tool in tools.tools:
print(f" - {tool.name}")
# Test a tool
result = await session.call_tool("get_today_date", {})
print(f"✅ Tool result: {result.content[0].text}")
except Exception as e:
print(f"❌ MCP Client Error: {e}")
traceback.print_exc()
async def main():
"""Main test function."""
print("🧪 Testing OAuth-enabled MCP server on AgentCore Runtime")
print("=" * 60)
# Test HTTP first
await test_http_first()
# Test MCP client
await test_mcp_client()
if __name__ == "__main__":
asyncio.run(main())
