05_http_client_demo.py

# HTTP MCP Client Demo with LLM Integration
# This demonstrates connecting to an HTTP MCP server and letting an LLM use its tools

import asyncio
import json
import aiohttp
from openai import OpenAI


async def main():
    """
    Demo showing how to connect to an HTTP MCP server via Python
    and interact with its tools and resources with LLM integration.
    """

    server_url = "http://localhost:8000/mcp"

    print("🌐 Connecting to HTTP MCP Server...")
    print(f"📍 Server URL: {server_url}")
    print("=" * 50)

    async with aiohttp.ClientSession() as session:

        # 1. Initialize MCP connection
        print("🤝 Initializing MCP connection...")
        init_request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"roots": {"listChanged": True}},
                "clientInfo": {"name": "HTTP Demo Client", "version": "1.0.0"},
            },
        }

        async with session.post(server_url, json=init_request) as response:
            if response.status != 200:
                print(f"❌ Failed to connect: HTTP {response.status}")
                return

            result = await response.json()
            print(f"✅ Connected! Server: {result['result']['serverInfo']['name']}")
            print()

        # 2. List available tools
        print("🔧 Discovering available tools...")
        tools_request = {
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/list",
            "params": {},
        }

        async with session.post(server_url, json=tools_request) as response:
            tools_result = await response.json()
            tools = tools_result["result"]["tools"]

            print(f"📋 Available tools ({len(tools)}):")
            for tool in tools:
                print(f"  • {tool['name']}: {tool['description']}")
            print()

        # 3. List available resources
        print("📂 Discovering available resources...")
        resources_request = {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "resources/list",
            "params": {},
        }

        try:
            async with session.post(server_url, json=resources_request) as response:
                resources_result = await response.json()
                if "result" in resources_result:
                    resources = resources_result["result"]["resources"]
                    print(f"📂 Available resources ({len(resources)}):")
                    for resource in resources:
                        print(f"  • {resource['uri']}: {resource['name']}")
                else:
                    print("📂 No resources available")
        except Exception as e:
            print(f"📂 No resources available: {e}")
        print()

        # 4. Demo: Let LLM use HTTP MCP tools
        if tools:
            print("🤖 Letting LLM use HTTP MCP tools...")

            # Convert MCP tools to OpenAI tool format
            openai_tools = []
            for tool in tools:
                openai_tools.append(
                    {
                        "type": "function",
                        "function": {
                            "name": tool["name"],
                            "description": tool["description"],
                            "parameters": tool.get(
                                "inputSchema", {"type": "object", "properties": {}}
                            ),
                        },
                    }
                )

            # Ask LLM to use the HTTP MCP server tools
            client = OpenAI()  # Make sure OPENAI_API_KEY is set

            try:
                response = client.chat.completions.create(
                    model="gpt-4o",
                    messages=[
                        {
                            "role": "user",
                            "content": "Get the current server time and start monitoring with a 3 second interval. Use session ID 'llm-demo-session'.",
                        }
                    ],
                    tools=openai_tools,
                    tool_choice="auto",
                )

                # Execute any tool calls the LLM wants to make
                message = response.choices[0].message
                if message.tool_calls:
                    print(f"🔧 LLM wants to use {len(message.tool_calls)} tool(s):")

                    for i, tool_call in enumerate(message.tool_calls):
                        print(f"   • {tool_call.function.name}")

                        # Execute the tool call via HTTP MCP
                        arguments = json.loads(tool_call.function.arguments)

                        mcp_request = {
                            "jsonrpc": "2.0",
                            "id": 10 + i,  # Unique ID for each tool call
                            "method": "tools/call",
                            "params": {
                                "name": tool_call.function.name,
                                "arguments": arguments,
                            },
                        }

                        async with session.post(
                            server_url, json=mcp_request
                        ) as response:
                            result = await response.json()

                            if "result" in result:
                                content = result["result"]["content"][0]["text"]
                                print(f"     ✅ Result: {content}")
                            else:
                                error_msg = result.get("error", {}).get(
                                    "message", "Unknown error"
                                )
                                print(f"     ❌ Error: {error_msg}")
                else:
                    print(f"💬 LLM response: {message.content}")

            except Exception as e:
                print(f"❌ LLM integration failed: {e}")
                print("   (Make sure OPENAI_API_KEY is set)")

                # Fallback to direct HTTP MCP usage
                print("\\n📝 Falling back to direct HTTP MCP usage...")

                # Test get_server_time tool directly
                direct_request = {
                    "jsonrpc": "2.0",
                    "id": 20,
                    "method": "tools/call",
                    "params": {"name": "get_server_time", "arguments": {}},
                }

                async with session.post(server_url, json=direct_request) as response:
                    result = await response.json()
                    if "result" in result:
                        content = result["result"]["content"][0]["text"]
                        print(f"   ⏰ Server time: {content}")

            print()

        # 5. Demo: Read a resource via HTTP
        if "resources" in locals() and resources:
            print("📖 Reading a resource via HTTP...")

            # Read the first available resource
            resource_uri = resources[0]["uri"]
            read_request = {
                "jsonrpc": "2.0",
                "id": 30,
                "method": "resources/read",
                "params": {"uri": resource_uri},
            }

            async with session.post(server_url, json=read_request) as response:
                result = await response.json()
                if "result" in result:
                    content = result["result"]["contents"][0]["text"]
                    print(f"   📄 Resource content: {content[:200]}...")
                else:
                    error_msg = result.get("error", {}).get("message", "Unknown error")
                    print(f"   ❌ Error reading resource: {error_msg}")
            print()

    print("✅ HTTP MCP Demo completed!")
    print()
    print("🔗 Key differences from stdio MCP:")
    print("   • Uses HTTP POST requests instead of process communication")
    print("   • Server runs independently (can be remote)")
    print("   • Multiple clients can connect simultaneously")
    print("   • Better for distributed systems and web deployments")


if __name__ == "__main__":
    print("🌐 HTTP MCP Client Demo")
    print("=" * 40)
    print("This demo connects to a running HTTP MCP server")
    print("Make sure your server is running:")
    print("  uvicorn 04_https_streamable:app --reload")
    print()

    asyncio.run(main())