1#!/usr/bin/env python3
2import argparse
3import json
4import sys
5import urllib.request
6 
7DEFAULT_URL = "https://mcp.sendpulse.com/mcp"
8DEFAULT_PROTOCOL_VERSION = "2025-03-26"
9 
10 
11def post_json(url: str, payload: dict, headers: dict):
12    data = json.dumps(payload).encode("utf-8")
13    req = urllib.request.Request(url, data=data, method="POST")
14    req.add_header("Content-Type", "application/json")
15    req.add_header("Accept", "application/json, text/event-stream")
16    for k, v in headers.items():
17        req.add_header(k, v)
18    with urllib.request.urlopen(req, timeout=30) as resp:
19        raw = resp.read().decode("utf-8")
20        return resp.status, dict(resp.headers.items()), raw
21 
22 
23def main():
24    parser = argparse.ArgumentParser(description="Test SendPulse MCP auth and basic protocol flow.")
25    parser.add_argument("--sp-id", help="SendPulse MCP/OpenAI ID header value")
26    parser.add_argument("--sp-secret", help="SendPulse MCP/OpenAI Secret header value")
27    parser.add_argument("--url", default=DEFAULT_URL, help="MCP URL")
28    parser.add_argument("--protocol-version", default=DEFAULT_PROTOCOL_VERSION, help="MCP protocol version")
29    parser.add_argument("--full-tools", action="store_true", help="Print full tools payload instead of summary")
30    args = parser.parse_args()
31 
32    sp_id = args.sp_id
33    sp_secret = args.sp_secret
34    if not sp_id or not sp_secret:
35        print("Missing --sp-id or --sp-secret", file=sys.stderr)
36        sys.exit(2)
37 
38    common_headers = {
39        "X-SP-ID": sp_id,
40        "X-SP-SECRET": sp_secret,
41    }
42 
43    initialize_payload = {
44        "jsonrpc": "2.0",
45        "id": 1,
46        "method": "initialize",
47        "params": {
48            "protocolVersion": args.protocol_version,
49            "capabilities": {},
50            "clientInfo": {"name": "sendpulse-mcp-handshake-test", "version": "1.0"},
51        },
52    }
53    status, headers, body = post_json(args.url, initialize_payload, common_headers)
54    init_data = json.loads(body)
55    session_id = headers.get("Mcp-Session-Id") or headers.get("mcp-session-id")
56    if status != 200 or not session_id:
57        print(json.dumps({
58            "ok": False,
59            "step": "initialize",
60            "status": status,
61            "hasSessionId": bool(session_id),
62            "body": init_data,
63        }, indent=2))
64        sys.exit(1)
65 
66    notify_payload = {
67        "jsonrpc": "2.0",
68        "method": "notifications/initialized",
69        "params": {},
70    }
71    notify_headers = dict(common_headers)
72    notify_headers["Mcp-Session-Id"] = session_id
73    notify_status, _, notify_body = post_json(args.url, notify_payload, notify_headers)
74    try:
75        notify_data = json.loads(notify_body)
76    except Exception:
77        notify_data = {"raw": notify_body}
78 
79    tools_payload = {
80        "jsonrpc": "2.0",
81        "id": 2,
82        "method": "tools/list",
83        "params": {},
84    }
85    tools_status, _, tools_body = post_json(args.url, tools_payload, notify_headers)
86    tools_data = json.loads(tools_body)
87    tools = tools_data.get("result", {}).get("tools", [])
88 
89    result = {
90        "ok": True,
91        "url": args.url,
92        "initialize": {
93            "status": status,
94            "sessionId": session_id,
95            "serverInfo": init_data.get("result", {}).get("serverInfo", {}),
96            "protocolVersion": init_data.get("result", {}).get("protocolVersion"),
97            "capabilities": sorted((init_data.get("result", {}).get("capabilities") or {}).keys()),
98        },
99        "initializedNotification": {
100            "status": notify_status,
101            "body": notify_data,
102        },
103        "tools": {
104            "status": tools_status,
105            "count": len(tools),
106            "sample": [tool.get("name") for tool in tools[:10]],
107        },
108    }
109 
110    if args.full_tools:
111        result["tools"]["full"] = tools
112 
113    print(json.dumps(result, indent=2))
114 
115 
116if __name__ == "__main__":
117    main()
118