"""
Test script for MCP Gateway
Tests the gateway functionality with example tools.
Note: In production (Lambda deployment), tools are automatically registered.
This script is for local testing and development.
"""
import asyncio
import json
import sys
import os
# Add current directory to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from gateway.gateway import MCPGateway
async def test_gateway():
"""Test MCP Gateway functionality"""
print("=" * 60)
print("MCP Gateway POC Test")
print("=" * 60)
# Initialize gateway
gateway = MCPGateway()
# Register example tools (using mock ARNs for testing)
print("\n1. Registering tools...")
# Register Lambda tool (mock)
gateway.register_lambda_tool(
name="weather_lookup",
lambda_arn="arn:aws:lambda:us-east-1:123456789012:function:weather-tool",
description="Get weather information for a location",
input_schema={
"type": "object",
"properties": {
"location": {"type": "string", "description": "City name"},
"units": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
},
"required": ["location"]
}
)
# Register REST tool
gateway.register_rest_tool(
name="stock_price",
endpoint="https://api.example.com/stock/price",
description="Get stock price information",
input_schema={
"type": "object",
"properties": {
"symbol": {"type": "string", "description": "Stock symbol"}
},
"required": ["symbol"]
},
method="POST"
)
print(f" Registered {len(gateway.tool_registry.list_tools())} tools")
# Test 1: Initialize
print("\n2. Testing initialize...")
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
init_response = await gateway.handle_request(init_request)
print(f" Response: {json.dumps(init_response, indent=2)}")
# Test 2: List tools
print("\n3. Testing tools/list...")
list_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
list_response = await gateway.handle_request(list_request)
print(f" Found {len(list_response.get('result', {}).get('tools', []))} tools")
for tool in list_response.get('result', {}).get('tools', []):
print(f" - {tool['name']}: {tool['description']}")
# Test 3: Call tool (this will fail without actual Lambda, but shows the flow)
print("\n4. Testing tools/call (weather_lookup)...")
call_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "weather_lookup",
"arguments": {
"location": "New York, NY",
"units": "celsius"
}
}
}
try:
call_response = await gateway.handle_request(call_request)
print(f" Response: {json.dumps(call_response, indent=2)}")
except Exception as e:
print(f" Expected error (no actual Lambda): {e}")
# Test 4: Unknown tool
print("\n5. Testing unknown tool...")
unknown_request = {
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "unknown_tool",
"arguments": {}
}
}
unknown_response = await gateway.handle_request(unknown_request)
print(f" Response: {json.dumps(unknown_response, indent=2)}")
print("\n" + "=" * 60)
print("Test completed!")
print("=" * 60)
if __name__ == "__main__":
asyncio.run(test_gateway())
