43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
import requests
|
|
import json
|
|
import sseclient
|
|
import threading
|
|
import sys
|
|
|
|
def main():
|
|
url = "https://axhubmcp.devjun.net/mcp/custom/cmm"
|
|
|
|
# 1. Start SSE connection
|
|
response = requests.get(url, stream=True, verify=False)
|
|
client = sseclient.SSEClient(response)
|
|
|
|
post_url = None
|
|
|
|
# We will read events in a background thread to not block the main thread
|
|
for event in client.events():
|
|
if event.event == 'endpoint':
|
|
# 2. Get POST endpoint
|
|
post_url = event.data
|
|
if not post_url.startswith('http'):
|
|
# Assuming relative URL, construct absolute
|
|
from urllib.parse import urljoin
|
|
post_url = urljoin(url, post_url)
|
|
|
|
# Send tools/list request
|
|
payload = {
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "tools/list"
|
|
}
|
|
requests.post(post_url, json=payload, verify=False)
|
|
|
|
elif event.event == 'message':
|
|
data = json.loads(event.data)
|
|
if data.get('id') == 1:
|
|
# 3. Print tools/list response
|
|
print(json.dumps(data, indent=2, ensure_ascii=False))
|
|
sys.exit(0)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|