darman
fe6d00ce6d
Adds an rmcp-based stdio MCP client in harness-mcp plus adapters exposing a configured server's tools as namespaced harness-tools, with an echo-server fixture and integration test, and wiring through harness-app.
87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Minimal MCP stdio server fixture for harness-mcp integration tests.
|
|
|
|
Speaks newline-delimited JSON-RPC (the framing rmcp's child-process transport uses) and
|
|
implements just enough of the protocol to be discovered and called: `initialize`,
|
|
`notifications/initialized`, `tools/list`, and `tools/call`. Exposes one tool, `echo`,
|
|
which returns its `text` argument, plus `boom`, which returns an error result.
|
|
"""
|
|
import json
|
|
import sys
|
|
|
|
PROTOCOL_VERSION = "2024-11-05"
|
|
|
|
TOOLS = [
|
|
{
|
|
"name": "echo",
|
|
"description": "Returns the text it is given.",
|
|
"inputSchema": {
|
|
"type": "object",
|
|
"properties": {"text": {"type": "string"}},
|
|
"required": ["text"],
|
|
},
|
|
},
|
|
{
|
|
"name": "boom",
|
|
"description": "Always fails.",
|
|
"inputSchema": {"type": "object", "properties": {}},
|
|
},
|
|
]
|
|
|
|
|
|
def reply(msg_id, result):
|
|
sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": msg_id, "result": result}) + "\n")
|
|
sys.stdout.flush()
|
|
|
|
|
|
def main():
|
|
# readline() rather than `for line in sys.stdin`: the latter's read-ahead buffer blocks
|
|
# until it fills, which would stall the JSON-RPC handshake line-by-line.
|
|
while True:
|
|
line = sys.stdin.readline()
|
|
if line == "": # EOF: parent closed stdin
|
|
break
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
msg = json.loads(line)
|
|
method = msg.get("method")
|
|
msg_id = msg.get("id")
|
|
|
|
if method == "initialize":
|
|
reply(msg_id, {
|
|
"protocolVersion": PROTOCOL_VERSION,
|
|
"capabilities": {"tools": {}},
|
|
"serverInfo": {"name": "echo-fixture", "version": "0.1.0"},
|
|
})
|
|
elif method == "notifications/initialized":
|
|
pass # notification: no response
|
|
elif method == "tools/list":
|
|
reply(msg_id, {"tools": TOOLS})
|
|
elif method == "tools/call":
|
|
params = msg.get("params") or {}
|
|
name = params.get("name")
|
|
args = params.get("arguments") or {}
|
|
if name == "echo":
|
|
reply(msg_id, {
|
|
"content": [{"type": "text", "text": args.get("text", "")}],
|
|
"isError": False,
|
|
})
|
|
elif name == "boom":
|
|
reply(msg_id, {
|
|
"content": [{"type": "text", "text": "kaboom"}],
|
|
"isError": True,
|
|
})
|
|
else:
|
|
reply(msg_id, {
|
|
"content": [{"type": "text", "text": f"unknown tool {name}"}],
|
|
"isError": True,
|
|
})
|
|
elif msg_id is not None:
|
|
# Unknown request: empty result keeps the client happy.
|
|
reply(msg_id, {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|