Impact Analytics

import httpx, json

# ─────────────────────────────────────────────────────────────────
# MCP Streamable HTTP client (JSON-RPC 2.0)
# Works with LangGraph, CrewAI, AutoGen, LlamaIndex, Haystack,
# or any custom agentic framework.
# ─────────────────────────────────────────────────────────────────

MCP_URL = "http://localhost:8008/mcp"
API_KEY = "YOUR_CLIENT_API_KEY"
HEADERS = {"X-API-Key": API_KEY, "Content-Type": "application/json"}

_req_id = 0

def _rpc(method: str, params: dict = {}) -> dict:
    global _req_id
    _req_id += 1
    resp = httpx.post(MCP_URL, headers=HEADERS, json={
        "jsonrpc": "2.0", "id": _req_id,
        "method": method, "params": params,
    })
    resp.raise_for_status()
    data = resp.json()
    if "error" in data:
        raise RuntimeError(data["error"]["message"])
    return data.get("result", {})


def initialize():
    """Perform MCP handshake."""
    return _rpc("initialize", {
        "protocolVersion": "2024-11-05",
        "clientInfo": {"name": "my-agent", "version": "1.0.0"},
        "capabilities": {},
    })


def list_tools() -> list[dict]:
    """Discover all tools available to this client."""
    return _rpc("tools/list").get("tools", [])


def call_tool(tool_name: str, arguments: dict = {}) -> str:
    """Execute a named tool with the given arguments."""
    result = _rpc("tools/call", {"name": tool_name, "arguments": arguments})
    content = result.get("content", [])
    return content[0]["text"] if content else ""


# ── Example usage ────────────────────────────────────────────────
if __name__ == "__main__":
    # 1. Handshake
    info = initialize()
    print("Connected to:", info["serverInfo"])

    # 2. Discover tools
    tools = list_tools()
    print("Available tools:", [t["name"] for t in tools])

    # 3. Call a tool
    result = call_tool("calculator", {"expression": "(100 + 50) * 0.2"})
    print("Calc:", result)

    # 4. Register as tools in your framework of choice:
    #    - LangGraph:  @tool decorator wrapping call_tool()
    #    - CrewAI:     Tool(name=..., func=call_tool)
    #    - AutoGen:    register_function(call_tool, ...)
    #    - LlamaIndex: FunctionTool.from_defaults(fn=call_tool)