75 lines
2.7 KiB
Python
Executable File
75 lines
2.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""最小 MCP stdio 服务器（宿主提供工具的对齐试验用）。
|
||
|
||
只给两个工具，不依赖任何第三方包：
|
||
echo_fact：返回固定事实字符串 mcp-fact-42
|
||
add：返回 a + b
|
||
|
||
协议：JSON-RPC over stdio，MCP initialize / tools/list / tools/call，
|
||
以及 notifications/initialized、ping。codex 的 MCP 客户端就是按
|
||
这个协议调它的，宿主只要在 config.toml 里写：
|
||
|
||
[mcp_servers.echo]
|
||
command = "python3"
|
||
args = ["<本文件绝对路径>"]
|
||
startup_timeout_sec = 30
|
||
"""
|
||
import json
|
||
import sys
|
||
|
||
|
||
def send(obj):
|
||
sys.stdout.write(json.dumps(obj) + "\n")
|
||
sys.stdout.flush()
|
||
|
||
|
||
def main():
|
||
for line in sys.stdin:
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
req = json.loads(line)
|
||
except Exception:
|
||
continue
|
||
method = req.get("method")
|
||
rid = req.get("id")
|
||
# 通知：没有 id，不回
|
||
if method in ("notifications/initialized", "notifications/cancelled"):
|
||
continue
|
||
if method == "initialize":
|
||
send({"jsonrpc": "2.0", "id": rid, "result": {
|
||
"protocolVersion": "2024-11-05",
|
||
"capabilities": {"tools": {}},
|
||
"serverInfo": {"name": "echo", "version": "0.1"}}})
|
||
elif method == "tools/list":
|
||
send({"jsonrpc": "2.0", "id": rid, "result": {"tools": [
|
||
{"name": "echo_fact",
|
||
"description": "返回一个固定事实字符串，用于验证宿主提供的 MCP 工具是否被 agent 调用",
|
||
"inputSchema": {"type": "object", "properties": {}}},
|
||
{"name": "add",
|
||
"description": "返回 a + b",
|
||
"inputSchema": {"type": "object",
|
||
"properties": {"a": {"type": "number"},
|
||
"b": {"type": "number"}},
|
||
"required": ["a", "b"]}}]}})
|
||
elif method == "tools/call":
|
||
params = req.get("params") or {}
|
||
name = params.get("name")
|
||
args = params.get("arguments") or {}
|
||
if name == "echo_fact":
|
||
text = "mcp-fact-42"
|
||
elif name == "add":
|
||
text = str(args.get("a", 0) + args.get("b", 0))
|
||
else:
|
||
text = f"unknown tool: {name}"
|
||
send({"jsonrpc": "2.0", "id": rid, "result": {
|
||
"content": [{"type": "text", "text": text}]}})
|
||
elif method == "ping":
|
||
send({"jsonrpc": "2.0", "id": rid, "result": {}})
|
||
elif rid is not None:
|
||
send({"jsonrpc": "2.0", "id": rid, "result": {}})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|