#!/usr/bin/env python
"""
Test script for SSE connection to the Sequential Thinking MCP server.
This script tests the SSE transport functionality.
"""
import asyncio
import json
import sys
import aiohttp
from typing import Optional
async def test_sse_connection(server_url: str = "http://localhost:8000"):
"""Test SSE connection to the MCP server."""
print(f"Testing SSE connection to: {server_url}")
# Test SSE endpoint
sse_url = f"{server_url}/sse"
message_url = f"{server_url}/messages/"
print(f"SSE endpoint: {sse_url}")
print(f"Message endpoint: {message_url}")
try:
# Test basic connectivity
async with aiohttp.ClientSession() as session:
# Test if server is running
async with session.get(server_url) as response:
if response.status == 200:
print("✓ Server is running")
else:
print(f"✗ Server returned status: {response.status}")
return False
# Test SSE endpoint
async with session.get(sse_url) as response:
if response.status == 200:
print("✓ SSE endpoint is accessible")
content_type = response.headers.get('Content-Type', '')
if 'text/event-stream' in content_type:
print("✓ Correct Content-Type for SSE")
else:
print(f"⚠ Unexpected Content-Type: {content_type}")
else:
print(f"✗ SSE endpoint returned status: {response.status}")
return False
# Test message endpoint
test_message = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "test-client",
"version": "1.0.0"
}
}
}
async with session.post(message_url, json=test_message) as response:
if response.status in [200, 204]:
print("✓ Message endpoint is accessible")
try:
result = await response.json()
print(f"✓ Received response: {json.dumps(result, indent=2)}")
except:
print("✓ Message sent successfully (no JSON response expected)")
else:
print(f"✗ Message endpoint returned status: {response.status}")
return False
except aiohttp.ClientConnectorError as e:
print(f"✗ Connection failed: {e}")
return False
except Exception as e:
print(f"✗ Unexpected error: {e}")
return False
print("\n🎉 SSE connection test completed successfully!")
return True
async def test_tool_invocation(server_url: str = "http://localhost:8000"):
"""Test tool invocation via SSE."""
print(f"\nTesting tool invocation via SSE...")
message_url = f"{server_url}/messages/"
# Test process_thought tool
test_tool_call = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "process_thought",
"arguments": {
"thought": "This is a test thought for SSE transport",
"thought_number": 1,
"total_thoughts": 1,
"next_thought_needed": False,
"stage": "Problem Definition",
"tags": ["test", "sse"],
"axioms_used": ["Testing is important"],
"assumptions_challenged": ["SSE transport works"]
}
}
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(message_url, json=test_tool_call) as response:
if response.status in [200, 204]:
print("✓ Tool invocation request sent successfully")
try:
result = await response.json()
print(f"✓ Tool response: {json.dumps(result, indent=2)}")
except:
print("✓ Tool invocation completed (no JSON response expected)")
else:
print(f"✗ Tool invocation failed with status: {response.status}")
return False
except Exception as e:
print(f"✗ Tool invocation error: {e}")
return False
print("🎉 Tool invocation test completed successfully!")
return True
def main():
"""Main function to run SSE tests."""
import argparse
parser = argparse.ArgumentParser(description="Test SSE connection to Sequential Thinking MCP Server")
parser.add_argument(
"--url",
default="http://localhost:8000",
help="Server URL (default: http://localhost:8000)"
)
parser.add_argument(
"--skip-tools",
action="store_true",
help="Skip tool invocation tests"
)
args = parser.parse_args()
print("🧪 Testing Sequential Thinking MCP Server SSE Transport")
print("=" * 60)
# Run connection test
connection_success = asyncio.run(test_sse_connection(args.url))
if not connection_success:
print("\n❌ Connection test failed. Please ensure the server is running with SSE transport.")
print("To start the server in SSE mode, run:")
print(" python run_sse_server.py --port 8000")
sys.exit(1)
# Run tool test if not skipped
if not args.skip_tools:
tool_success = asyncio.run(test_tool_invocation(args.url))
if not tool_success:
print("\n❌ Tool invocation test failed.")
sys.exit(1)
print("\n✅ All tests passed!")
print("\n📋 Usage instructions:")
print(f"1. Server URL: {args.url}")
print("2. SSE endpoint: /sse")
print("3. Message endpoint: /messages/")
print("4. Available tools: process_thought, generate_summary, clear_history, export_session, import_session")
if __name__ == "__main__":
main()
