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.
73 lines
2.4 KiB
73 lines
2.4 KiB
4 weeks ago
|
import sys, json, urllib.request
|
||
|
|
|
||
|
|
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=1200):
|
||
|
|
h = dict(HEADERS)
|
||
|
|
if session:
|
||
|
|
h["mcp-session-id"] = session
|
||
|
|
data = json.dumps(body).encode()
|
||
|
|
req = urllib.request.Request(URL, data=data, headers=h, method="POST")
|
||
|
|
resp = urllib.request.urlopen(req, timeout=timeout)
|
||
|
|
sid = resp.headers.get("mcp-session-id", session)
|
||
|
|
# read SSE stream
|
||
|
|
result = None
|
||
|
|
for raw in resp:
|
||
|
|
line = raw.decode("utf-8", "replace").strip()
|
||
|
|
if not line or line.startswith(":"):
|
||
|
|
continue
|
||
|
|
if line.startswith("data:"):
|
||
|
|
payload = line[5:].strip()
|
||
|
|
try:
|
||
|
|
d = json.loads(payload)
|
||
|
|
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]
|
||
|
|
args = json.loads(sys.argv[2])
|
||
|
|
outfile = sys.argv[3] if len(sys.argv) > 3 else None
|
||
|
|
|
||
|
|
# init
|
||
|
|
sid, _ = post({"jsonrpc":"2.0","id":1,"method":"initialize",
|
||
|
|
"params":{"protocolVersion":"2024-11-05","capabilities":{},
|
||
|
|
"clientInfo":{"name":"pi","version":"1.0"}}}, timeout=30)
|
||
|
|
# initialized notification (no response expected)
|
||
|
|
try:
|
||
|
|
h = dict(HEADERS); h["mcp-session-id"] = sid
|
||
|
|
req = urllib.request.Request(URL,
|
||
|
|
data=json.dumps({"jsonrpc":"2.0","method":"notifications/initialized"}).encode(),
|
||
|
|
headers=h, method="POST")
|
||
|
|
urllib.request.urlopen(req, timeout=15).read()
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
# call tool
|
||
|
|
_, res = post({"jsonrpc":"2.0","id":2,"method":"tools/call",
|
||
|
|
"params":{"name":tool,"arguments":args}}, session=sid, timeout=1800)
|
||
|
|
if res is None:
|
||
|
|
print("NO_RESULT")
|
||
|
|
return
|
||
|
|
if "error" in res:
|
||
|
|
print("ERROR:", json.dumps(res["error"], ensure_ascii=False))
|
||
|
|
return
|
||
|
|
texts = []
|
||
|
|
for c in res["result"].get("content", []):
|
||
|
|
if c.get("type") == "text":
|
||
|
|
texts.append(c["text"])
|
||
|
|
output = "\n\n".join(texts)
|
||
|
|
if outfile:
|
||
|
|
open(outfile, "w", encoding="utf-8").write(output)
|
||
|
|
print("WROTE", len(output), "chars to", outfile)
|
||
|
|
else:
|
||
|
|
print(output)
|
||
|
|
|
||
|
|
main()
|