|
import json |
|
import sys |
|
from datetime import datetime |
|
|
|
def log(message): |
|
"""Log to stderr with timestamp""" |
|
timestamp = datetime.now().strftime("%H:%M:%S") |
|
print(f"[{timestamp}] {message}", file=sys.stderr, flush=True) |
|
|
|
def create_response(id, result): |
|
return {"jsonrpc": "2.0", "id": id, "result": result} |
|
|
|
def create_error(id, code, message): |
|
return {"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}} |
|
|
|
def handle_request(request): |
|
method = request.get("method") |
|
params = request.get("params", {}) |
|
request_id = request.get("id") |
|
|
|
log(f"Received: {method}") |
|
|
|
if method == "initialize": |
|
log("Initializing MCP server") |
|
return create_response(request_id, { |
|
"protocolVersion": "2024-11-05", |
|
"capabilities": {"tools": {}}, |
|
"serverInfo": {"name": "debug-mcp-server", "version": "1.0.0"} |
|
}) |
|
|
|
elif method == "notifications/initialized": |
|
log("Server initialized") |
|
return None |
|
|
|
elif method == "tools/list": |
|
return create_response(request_id, { |
|
"tools": [ |
|
{ |
|
"name": "test_mixed_params", |
|
"description": "Test exact n3-notes create_note schema pattern", |
|
"inputSchema": { |
|
"$schema": "http://json-schema.org/draft-07/schema#", |
|
"type": "object", |
|
"title": "TestMixedInput", |
|
"properties": { |
|
"content": {"type": "string"}, |
|
"parent": {"default": None, "type": ["string", "null"]} |
|
}, |
|
"required": ["content"] |
|
} |
|
}, |
|
{ |
|
"name": "test_all_required", |
|
"description": "Test all required parameters", |
|
"inputSchema": { |
|
"type": "object", |
|
"properties": { |
|
"field1": {"type": "string"}, |
|
"field2": {"type": "string"} |
|
}, |
|
"required": ["field1", "field2"] |
|
} |
|
}, |
|
{ |
|
"name": "test_all_optional", |
|
"description": "Test all optional parameters", |
|
"inputSchema": { |
|
"type": "object", |
|
"properties": { |
|
"opt1": {"type": "string", "default": "default1"}, |
|
"opt2": {"type": "string", "default": "default2"} |
|
}, |
|
"required": [] |
|
} |
|
} |
|
] |
|
}) |
|
|
|
elif method == "tools/call": |
|
tool_name = params.get("name") |
|
arguments = params.get("arguments", {}) |
|
|
|
log(f"=== TOOL CALL DEBUG ===") |
|
log(f"Tool: {tool_name}") |
|
log(f"Arguments received: {json.dumps(arguments)}") |
|
log(f"Arguments type: {type(arguments)}") |
|
log(f"Arguments empty: {len(arguments) == 0}") |
|
log("========================") |
|
|
|
if tool_name == "test_mixed_params": |
|
if "content" not in arguments: |
|
return create_error(request_id, -32602, "Missing required field 'content'") |
|
|
|
content = arguments["content"] |
|
parent = arguments.get("parent", None) |
|
result = f"✅ Mixed params worked! Content: '{content}', Parent: {parent}" |
|
|
|
elif tool_name == "test_all_required": |
|
if "field1" not in arguments or "field2" not in arguments: |
|
return create_error(request_id, -32602, "Missing required fields") |
|
result = f"✅ All required worked! field1: '{arguments['field1']}', field2: '{arguments['field2']}'" |
|
|
|
elif tool_name == "test_all_optional": |
|
opt1 = arguments.get("opt1", "default1") |
|
opt2 = arguments.get("opt2", "default2") |
|
result = f"✅ All optional worked! opt1: '{opt1}', opt2: '{opt2}'" |
|
|
|
else: |
|
return create_error(request_id, -32601, f"Unknown tool: {tool_name}") |
|
|
|
return create_response(request_id, {"content": [{"type": "text", "text": result}]}) |
|
|
|
return create_error(request_id, -32601, f"Unknown method: {method}") |
|
|
|
def main(): |
|
log("Starting debug MCP server...") |
|
|
|
try: |
|
for line in sys.stdin: |
|
line = line.strip() |
|
if not line: |
|
continue |
|
|
|
try: |
|
request = json.loads(line) |
|
response = handle_request(request) |
|
|
|
if response: |
|
print(json.dumps(response), flush=True) |
|
|
|
except json.JSONDecodeError as e: |
|
log(f"JSON decode error: {e}") |
|
except Exception as e: |
|
log(f"Error handling request: {e}") |
|
if "id" in locals(): |
|
error_response = create_error(request.get("id"), -32603, str(e)) |
|
print(json.dumps(error_response), flush=True) |
|
|
|
except KeyboardInterrupt: |
|
log("Server stopped") |
|
|
|
if __name__ == "__main__": |
|
main() |
