Files
pyMCPBroker/tests/fake_mcp_server.py

126 lines
3.7 KiB
Python

from __future__ import annotations
import json
import sys
TOOLS = [
{
"name": "get_file_contents",
"title": "Get file contents",
"description": "Read file contents at a given ref.",
"inputSchema": {
"type": "object",
"properties": {
"owner": {"type": "string"},
"repo": {"type": "string"},
"ref": {"type": "string"},
"filePath": {"type": "string"},
"withLines": {"type": "boolean"},
},
"required": ["owner", "repo", "ref", "filePath"],
"additionalProperties": False,
},
},
{
"name": "list_branches",
"description": "List all branches in a repository for a given owner and repo.",
"inputSchema": {
"type": "object",
"properties": {
"owner": {"type": "string"},
"repo": {"type": "string"}
},
"required": ["owner", "repo"],
"additionalProperties": False
}
},
{
"name": "delete_file",
"title": "Delete file",
"description": "Dangerous write tool.",
"inputSchema": {
"type": "object",
"properties": {"filePath": {"type": "string"}},
"required": ["filePath"],
},
},
]
def send(message: dict) -> None:
sys.stdout.write(json.dumps(message, separators=(",", ":")) + "\n")
sys.stdout.flush()
for line in sys.stdin:
raw = line.strip()
if not raw:
continue
msg = json.loads(raw)
method = msg.get("method")
msg_id = msg.get("id")
if method == "initialize":
send(
{
"jsonrpc": "2.0",
"id": msg_id,
"result": {
"protocolVersion": msg["params"]["protocolVersion"],
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "fake-mcp", "version": "1.0"},
},
}
)
continue
if method == "notifications/initialized":
continue
if method == "tools/list":
cursor = (msg.get("params") or {}).get("cursor")
if not cursor:
send(
{
"jsonrpc": "2.0",
"id": msg_id,
"result": {"tools": [TOOLS[0], TOOLS[1]], "nextCursor": "page-2"},
}
)
else:
send(
{
"jsonrpc": "2.0",
"id": msg_id,
"result": {"tools": [TOOLS[2]]},
}
)
continue
if method == "tools/call":
params = msg.get("params") or {}
name = params["name"]
arguments = params.get("arguments") or {}
if name == "get_file_contents":
payload = {
"content": [
{
"type": "text",
"text": "README line\n" * 3000,
}
],
"structuredContent": {
"owner": arguments.get("owner"),
"repo": arguments.get("repo"),
"ref": arguments.get("ref"),
"filePath": arguments.get("filePath"),
},
"isError": False,
}
send({"jsonrpc": "2.0", "id": msg_id, "result": payload})
continue
send({"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Unknown tool: {name}"}})
continue
send({"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Unknown method: {method}"}})