01_python_client_demo.py

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

import asyncio
import json
from contextlib import AsyncExitStack
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
from openai import OpenAI


async def main():
    """
    Simple demo showing how to connect to an MCP server via Python
    and interact with its tools and resources.
    """

    async with AsyncExitStack() as stack:
        # Connect to a local MCP server (filesystem server example)
        # In practice, this could be any MCP server: stdio, SSE, or HTTP
        server_params = StdioServerParameters(
            command="npx",
            args=[
                "-y",
                "@modelcontextprotocol/server-filesystem",
                "/Users/davidhague/Downloads",
            ],
        )
        transport = await stack.enter_async_context(stdio_client(server_params))

        # Create client session
        read_stream, write_stream = transport
        session = await stack.enter_async_context(
            ClientSession(read_stream, write_stream)
        )

        # Initialize connection
        await session.initialize()

        print("🚀 Connected to MCP server!")
        print("=" * 50)

        # 1. List available tools
        tools_result = await session.list_tools()
        print(f"📋 Available tools ({len(tools_result.tools)}):")
        for tool in tools_result.tools:
            print(f"  • {tool.name}: {tool.description}")
        print()

        # 2. List available resources
        try:
            resources_result = await session.list_resources()
            print(f"📂 Available resources ({len(resources_result.resources)}):")
            for resource in resources_result.resources:
                print(f"  • {resource.uri}: {resource.name}")
        except Exception as e:
            print(f"📂 No resources available: {e}")
        print()

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

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

            # Ask LLM to create and manage files
            client = OpenAI()  # Make sure OPENAI_API_KEY is set

            try:
                response = client.chat.completions.create(
                    model="gpt-4o",
                    messages=[
                        {
                            "role": "user",
                            "content": "Create a file called '/Users/davidhague/Downloads/llm_demo.txt' with a creative message about AI and MCP working together. Then read it back to confirm it was created.",
                        }
                    ],
                    tools=openai_tools,
                    tool_choice="auto",
                )
                print("\n--- LLM Request Sent ---")
                print(f"Messages: {response.choices[0].message.content}")
                print(f"Tool Calls: {response.choices[0].message.tool_calls}")
                print("--- End LLM Request ---")

                # Execute any tool calls the LLM wants to make
                message = response.choices[0].message
                print(f"\n--- LLM Response Received ---")
                print(f"Message: {message.content}")
                print(f"Tool Calls: {message.tool_calls}")
                print("--- End LLM Response ---")

                if message.tool_calls:
                    print(f"\n🔧 LLM wants to use {len(message.tool_calls)} tool(s):")

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

                        # Execute the tool call via MCP
                        arguments = json.loads(tool_call.function.arguments)
                        result = await session.call_tool(
                            tool_call.function.name, arguments=arguments
                        )

                        print(
                            f"     Result: {result.content[0].text if result.content else 'Success!'}"
                        )

                        # If it was a read operation, show some content
                        if tool_call.function.name == "read_file" and result.content:
                            content = result.content[0].text
                            preview = (
                                content[:150] + "..." if len(content) > 150 else content
                            )
                            print(f"     Content preview: {preview}")
                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 MCP tool usage
                print("\n📝 Falling back to direct MCP usage...")
                if any(tool.name == "write_file" for tool in tools_result.tools):
                    result = await session.call_tool(
                        "write_file",
                        arguments={
                            "path": "/Users/davidhague/Downloads/mcp_demo.txt",
                            "content": "Hello from MCP! This file was created via the Model Context Protocol.",
                        },
                    )
                    print(
                        f"   File created: {result.content[0].text if result.content else 'Success!'}"
                    )

            print()

        print("\n✅ MCP Demo completed!")


if __name__ == "__main__":
    asyncio.run(main())