137 lines
3.5 KiB
Python
137 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mock MCP server for integration testing.
|
|
Implements the MCP protocol (initialize, tools/list, tools/call).
|
|
"""
|
|
import json
|
|
import sys
|
|
import os
|
|
|
|
# Tools exposed by this mock server
|
|
TOOLS = [
|
|
{
|
|
"name": "test_tool",
|
|
"description": "A test tool that echoes its input",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"message": {"type": "string", "description": "Message to echo"}
|
|
},
|
|
"required": ["message"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_time",
|
|
"description": "Returns the current server time",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
}
|
|
]
|
|
|
|
|
|
def handle_initialize(params):
|
|
"""Handle initialize request"""
|
|
return {
|
|
"protocolVersion": "2024-11-05",
|
|
"capabilities": {
|
|
"tools": {}
|
|
},
|
|
"serverInfo": {
|
|
"name": "MockMCPServer",
|
|
"version": "1.0.0"
|
|
}
|
|
}
|
|
|
|
|
|
def handle_tools_list(params):
|
|
"""Handle tools/list request"""
|
|
return {"tools": TOOLS}
|
|
|
|
|
|
def handle_tools_call(params):
|
|
"""Handle tools/call request"""
|
|
tool_name = params.get("name", "")
|
|
arguments = params.get("arguments", {})
|
|
|
|
if tool_name == "test_tool":
|
|
message = arguments.get("message", "no message")
|
|
return {
|
|
"content": [
|
|
{"type": "text", "text": f"Echo: {message}"}
|
|
]
|
|
}
|
|
elif tool_name == "get_time":
|
|
import datetime
|
|
return {
|
|
"content": [
|
|
{"type": "text", "text": datetime.datetime.now().isoformat()}
|
|
]
|
|
}
|
|
else:
|
|
return {
|
|
"content": [
|
|
{"type": "text", "text": f"Unknown tool: {tool_name}"}
|
|
],
|
|
"isError": True
|
|
}
|
|
|
|
|
|
def handle_request(request):
|
|
"""Route request to appropriate handler"""
|
|
method = request.get("method", "")
|
|
|
|
handlers = {
|
|
"initialize": handle_initialize,
|
|
"tools/list": handle_tools_list,
|
|
"tools/call": handle_tools_call,
|
|
}
|
|
|
|
handler = handlers.get(method)
|
|
if handler:
|
|
return handler(request.get("params", {}))
|
|
else:
|
|
return {"error": {"code": -32601, "message": f"Method not found: {method}"}}
|
|
|
|
|
|
def main():
|
|
while True:
|
|
try:
|
|
line = sys.stdin.readline()
|
|
if not line:
|
|
break
|
|
|
|
request = json.loads(line.strip())
|
|
result = handle_request(request)
|
|
|
|
response = {
|
|
"jsonrpc": "2.0",
|
|
"id": request.get("id")
|
|
}
|
|
|
|
if "error" in result:
|
|
response["error"] = result["error"]
|
|
else:
|
|
response["result"] = result
|
|
|
|
sys.stdout.write(json.dumps(response) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
except json.JSONDecodeError as e:
|
|
error_response = {
|
|
"jsonrpc": "2.0",
|
|
"id": None,
|
|
"error": {"code": -32700, "message": f"Parse error: {str(e)}"}
|
|
}
|
|
sys.stdout.write(json.dumps(error_response) + "\n")
|
|
sys.stdout.flush()
|
|
except Exception as e:
|
|
# Log to stderr for debugging
|
|
sys.stderr.write(f"Error: {str(e)}\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|