60 lines
1.8 KiB
Python
60 lines
1.8 KiB
Python
import urllib.request
|
|
import json
|
|
import threading
|
|
import sys
|
|
import time
|
|
|
|
sse_url = "http://localhost:3000/sse"
|
|
|
|
def call_mcp(method, params):
|
|
req = urllib.request.Request(sse_url)
|
|
with urllib.request.urlopen(req) as response:
|
|
endpoint = None
|
|
# read the SSE stream
|
|
for line_bytes in response:
|
|
line = line_bytes.decode('utf-8').strip()
|
|
if line.startswith("data: /messages"):
|
|
endpoint = "http://localhost:3000" + line.split("data: ")[1].strip()
|
|
break
|
|
|
|
if not endpoint:
|
|
print("Failed to get endpoint")
|
|
sys.exit(1)
|
|
|
|
print(f"Got endpoint: {endpoint}")
|
|
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "tools/call",
|
|
"params": {
|
|
"name": method,
|
|
"arguments": params
|
|
}
|
|
}
|
|
|
|
post_req = urllib.request.Request(endpoint, data=json.dumps(payload).encode('utf-8'), headers={'Content-Type': 'application/json'})
|
|
with urllib.request.urlopen(post_req) as post_resp:
|
|
print(f"POST status: {post_resp.status}")
|
|
|
|
# continue reading SSE for response
|
|
for line_bytes in response:
|
|
line = line_bytes.decode('utf-8').strip()
|
|
if line.startswith("data: "):
|
|
data = line[6:]
|
|
if data == "ping": continue
|
|
try:
|
|
msg = json.loads(data)
|
|
if msg.get("id") == 1:
|
|
print("RESULT:")
|
|
print(json.dumps(msg, indent=2))
|
|
return
|
|
except:
|
|
pass
|
|
|
|
import sys
|
|
if __name__ == "__main__":
|
|
method = sys.argv[1]
|
|
params = json.loads(sys.argv[2])
|
|
call_mcp(method, params)
|