Simple HTTP MCP Client

mcp_client.py

#!/usr/bin/env python3

"""
Generic MCP client for communicating with Streamable HTTP MCP servers.

Very basic. Doesn't support any kind of auth.
No external dependencies.
Output is always JSON, unless there's an error.
DON'T use in an app or if performance matters.

Usage:
  ./mcp_client.py <mcp_url>                       # List available tools
  ./mcp_client.py <mcp_url> <tool> <json_input>   # Call a specific tool

"""

import json
import sys
import urllib.request
from typing import Any


def clean_null_values(obj):
    """Recursively remove null values and empty dictionaries from a data structure."""
    if isinstance(obj, dict):
        cleaned = {}
        for key, value in obj.items():
            if value is not None:
                cleaned_value = clean_null_values(value)
                if cleaned_value is not None and cleaned_value != {}:
                    cleaned[key] = cleaned_value
        return cleaned if cleaned else None
    elif isinstance(obj, list):
        cleaned = []
        for item in obj:
            cleaned_item = clean_null_values(item)
            if cleaned_item is not None and cleaned_item != {}:
                cleaned.append(cleaned_item)
        return cleaned if cleaned else None
    else:
        return obj


class MCPClient:
    def __init__(self, mcp_url: str):
        self.mcp_url = mcp_url
        self.opener = urllib.request.build_opener()
        self.session_id = None

    def initialize_session(self):
        """Initialize the MCP session."""
        if self.session_id:
            return

        # Proper MCP initialize request
        init_payload = {
            "jsonrpc": "2.0",
            "id": 0,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"experimental": {}, "sampling": {}},
                "clientInfo": {"name": "mcp-client", "version": "1.0.0"},
            },
        }

        headers = {
            "Accept": "text/event-stream, application/json",
            "Content-Type": "application/json",
        }

        try:
            response = self._make_request(self.mcp_url, init_payload, headers)

            # Extract session ID from headers
            if "mcp-session-id" in response.headers:
                self.session_id = response.headers["mcp-session-id"]

                # Send initialized notification to complete handshake
                initialized_payload = {
                    "jsonrpc": "2.0",
                    "method": "notifications/initialized",
                    "params": {},
                }

                init_headers = {
                    "Accept": "text/event-stream, application/json",
                    "Content-Type": "application/json",
                    "mcp-session-id": self.session_id,
                }

                # Send notification (no response expected)
                self._make_request(self.mcp_url, initialized_payload, init_headers)
        except Exception:
            pass  # Silent initialization

    def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        """Call a tool on the MCP server using SSE."""
        # Ensure session is initialized
        self.initialize_session()

        url = self.mcp_url

        payload = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {"name": tool_name, "arguments": arguments},
        }

        headers = {
            "Accept": "text/event-stream, application/json",
            "Content-Type": "application/json",
        }

        # Add session ID to headers if available
        if self.session_id:
            headers["mcp-session-id"] = self.session_id

        response = self._make_streaming_request(url, payload, headers)
        return response

    def list_tools(self) -> dict[str, Any]:
        """List all available tools on the MCP server."""
        # Ensure session is initialized
        self.initialize_session()

        payload = {"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}

        headers = {
            "Accept": "text/event-stream, application/json",
            "Content-Type": "application/json",
        }

        # Add session ID to headers if available
        if self.session_id:
            headers["mcp-session-id"] = self.session_id

        response = self._make_streaming_request(self.mcp_url, payload, headers)
        return response

    def _make_request(self, url: str, payload: dict, headers: dict):
        """Make a simple HTTP POST request."""
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(url, data=data, headers=headers)
        return self.opener.open(req)

    def _make_streaming_request(
        self, url: str, payload: dict, headers: dict
    ) -> dict[str, Any]:
        """Make a streaming HTTP POST request and parse SSE response."""
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request(url, data=data, headers=headers)
        response = self.opener.open(req)

        if response.status != 200:
            raise Exception(f"HTTP {response.status}: {response.reason}")

        result = None
        for line in response:
            line = line.decode("utf-8").strip()
            if line.startswith("data: "):
                data = line[6:]  # Remove "data: " prefix
                if data.strip() == "[DONE]":
                    break
                try:
                    result = json.loads(data)
                    if "error" in result:
                        raise Exception(f"MCP Error: {result['error']}")
                    return result.get("result", result)
                except json.JSONDecodeError:
                    continue

        if result is None:
            raise Exception("No valid response received from server")

        return result

    def close(self):
        """Close the HTTP client."""
        # urllib doesn't need explicit cleanup
        return


def main():
    args = sys.argv[1:]
    if not args:
        print(
            "Usage: python mcp_client.py <mcp_url> [tool_name] [json_input]",
            file=sys.stderr,
        )
        sys.exit(1)

    mcp_url = args[0]
    client = MCPClient(mcp_url)

    args = args[1:]

    try:
        if not args:
            # List all available tools (default action)
            result = client.list_tools()
            cleaned_result = clean_null_values(result)
            print(json.dumps(cleaned_result, indent=2))

        elif len(args) == 2:
            # Call a specific tool
            tool_name, json_input = args

            try:
                tool_args = json.loads(json_input)
            except json.JSONDecodeError as e:
                print(f"Error: Invalid JSON input: {e}", file=sys.stderr)
                sys.exit(1)

            result = client.call_tool(tool_name, tool_args)

            # Extract and clean the content
            if "content" in result and len(result["content"]) > 0:
                content_text = result["content"][0]["text"]
                try:
                    # Try to parse as JSON first
                    data = json.loads(content_text)
                    cleaned_data = clean_null_values(data)
                    print(json.dumps(cleaned_data, indent=2))
                except json.JSONDecodeError:
                    # If not JSON, print as plain text
                    print(content_text)
            else:
                cleaned_result = clean_null_values(result)
                print(json.dumps(cleaned_result, indent=2))

        else:
            print(
                "Usage: python mcp_client.py <mcp_url> [tool_name] [json_input]",
                file=sys.stderr,
            )
            sys.exit(1)

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)
    finally:
        client.close()


if __name__ == "__main__":
    main()