You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
67 lines
2.4 KiB
67 lines
2.4 KiB
"""蓝湖 MCP 通用调用脚本（wayfinder lead-docs-audit 配套工具）。
|
|
|
|
用法：
|
|
python tmp/lanhu_mcp.py <tool名> '<json参数>'
|
|
示例：
|
|
python tmp/lanhu_mcp.py lanhu_get_pages '{"url": "https://lanhuapp.com/..."}'
|
|
输出：UTF-8 JSON（data 行解析结果）。
|
|
"""
|
|
import json, sys, urllib.request
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
URL = "http://127.0.0.1:8000/mcp"
|
|
HEADERS = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"}
|
|
|
|
|
|
def post(body, session=None, timeout=180):
|
|
h = dict(HEADERS)
|
|
if session:
|
|
h["mcp-session-id"] = session
|
|
req = urllib.request.Request(URL, data=json.dumps(body).encode(), headers=h, method="POST")
|
|
resp = urllib.request.urlopen(req, timeout=timeout)
|
|
sid = resp.headers.get("mcp-session-id", session)
|
|
result = None
|
|
for raw in resp:
|
|
line = raw.decode("utf-8", "replace").strip()
|
|
if not line or line.startswith(":"):
|
|
continue
|
|
if line.startswith("data:"):
|
|
try:
|
|
d = json.loads(line[5:].strip())
|
|
except Exception:
|
|
continue
|
|
if isinstance(d, dict) and ("result" in d or "error" in d):
|
|
result = d
|
|
break
|
|
return sid, result
|
|
|
|
|
|
def main():
|
|
tool = sys.argv[1]
|
|
raw = sys.argv[2] if len(sys.argv) > 2 else "{}"
|
|
if raw.startswith("@"):
|
|
with open(raw[1:], "r", encoding="utf-8") as f:
|
|
raw = f.read()
|
|
args = json.loads(raw)
|
|
sid, _ = post({"jsonrpc": "2.0", "id": 1, "method": "initialize",
|
|
"params": {"protocolVersion": "2024-11-05", "capabilities": {},
|
|
"clientInfo": {"name": "cli", "version": "1.0"}}}, timeout=30)
|
|
h = dict(HEADERS)
|
|
h["mcp-session-id"] = sid
|
|
urllib.request.urlopen(urllib.request.Request(
|
|
URL, data=json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}).encode(),
|
|
headers=h, method="POST"), timeout=15).read()
|
|
_, res = post({"jsonrpc": "2.0", "id": 2, "method": "tools/call",
|
|
"params": {"name": tool, "arguments": args}}, session=sid)
|
|
out = json.dumps(res, ensure_ascii=False, indent=1)
|
|
if len(sys.argv) > 3:
|
|
with open(sys.argv[3], "w", encoding="utf-8") as f:
|
|
f.write(out)
|
|
print(f"written: {sys.argv[3]} ({len(out)} chars)")
|
|
else:
|
|
print(out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|