test_ai_via_mcp.py•4.39 kB
#!/usr/bin/env python3
"""
Test AI functionality via MCP protocol with proper session handling.
"""
import asyncio
import json
import aiohttp
import sys
async def test_ai_via_mcp():
"""Test AI tools via MCP protocol."""
print("🤖 Testing AI Tools via MCP Protocol")
print("="*50)
server_url = "http://127.0.0.1:8888/mcp/"
async with aiohttp.ClientSession() as session:
# 1. Initialize MCP session
print("1️⃣ Initializing MCP session...")
init_request = {
"jsonrpc": "2.0",
"id": "1",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "ai-test-client", "version": "1.0.0"}
}
}
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
try:
async with session.post(server_url, json=init_request, headers=headers) as resp:
if resp.content_type == 'text/plain':
# Handle Server-Sent Events
content = await resp.text()
print(f" MCP Session Response: {content[:100]}...")
else:
result = await resp.json()
print(f" MCP Session: {result.get('result', {}).get('serverInfo', {}).get('name', 'Unknown')}")
except Exception as e:
print(f" MCP Session Error: {e}")
return
# 2. List available tools
print("\n2️⃣ Listing available AI tools...")
list_tools_request = {
"jsonrpc": "2.0",
"id": "2",
"method": "tools/list",
"params": {}
}
try:
async with session.post(server_url, json=list_tools_request, headers=headers) as resp:
if resp.content_type.startswith('text/'):
content = await resp.text()
# Parse SSE format
if "data:" in content:
data_line = [line for line in content.split('\n') if line.startswith('data:')]
if data_line:
data = json.loads(data_line[0][5:]) # Remove 'data:' prefix
tools = data.get('result', {}).get('tools', [])
ai_tools = [t for t in tools if 'ai' in t.get('name', '').lower()]
print(f" Found {len(ai_tools)} AI tools:")
for tool in ai_tools:
print(f" • {tool.get('name')}: {tool.get('description', '')[:60]}...")
except Exception as e:
print(f" Tools listing error: {e}")
# 3. Test AI system status
print("\n3️⃣ Testing AI system status...")
status_request = {
"jsonrpc": "2.0",
"id": "3",
"method": "tools/call",
"params": {
"name": "get_ai_system_status",
"arguments": {}
}
}
try:
async with session.post(server_url, json=status_request, headers=headers) as resp:
content = await resp.text()
if "data:" in content:
data_lines = [line for line in content.split('\n') if line.startswith('data:')]
for line in data_lines:
try:
data = json.loads(line[5:])
if 'result' in data:
result_content = data['result']['content'][0]['text']
print(f" AI Status: {result_content[:200]}...")
break
except:
continue
except Exception as e:
print(f" AI status error: {e}")
print(f"\n✅ MCP AI Test Summary:")
print(f" • Server: Running at {server_url}")
print(f" • Protocol: MCP 2024-11-05")
print(f" • Transport: Server-Sent Events")
print(f" • AI Environment: production")
print(f" • Local Model: phi3 via Ollama")
if __name__ == "__main__":
asyncio.run(test_ai_via_mcp())