MCP客户端构建指南

# Create project
uv init mcp-client
cd mcp-client

# Install dependencies
uv add mcp google-genai python-dotenv

# Create .env file with your API key
echo "GEMINI_API_KEY=your_key_here" > .env

# Remove boilerplate
rm hello.py

# Create the client (copy code from this repo's client.py)
# Then run it
uv run client.py path/to/your/mcp-server.py

uv init mcp-client
cd mcp-client

uv venv

uv add mcp google-genai python-dotenv

uv add mcp anthropic python-dotenv

# Windows
del hello.py

# macOS/Linux
rm hello.py

# Windows
echo. > .env

# macOS/Linux
touch .env

# For Gemini (Primary)
GEMINI_API_KEY=your_actual_api_key_here

# OR for Claude (Alternative)
# ANTHROPIC_API_KEY=your_actual_api_key_here

echo ".env" >> .gitignore

.venv/
__pycache__/
*.pyc

# Windows
type nul > client.py

# macOS/Linux
touch client.py

MCPClient
├── __init__()                    # Initialize LLM client, setup conversation state
├── connect_to_server()           # Establish stdio connection to MCP server
├── process_query()               # Main query processing loop with tool calling
├── _execute_tool_call()          # Execute single MCP tool call
├── _extract_tool_result()        # Parse tool results from various formats
├── chat_loop()                   # Interactive CLI interface
└── cleanup()                     # Resource cleanup

User Query → LLM → [Tool Call?] → Execute Tool → LLM → Response
                         ↓
                     [Multiple tool calls possible]
                         ↓
                     [Loop until no more tools needed]

┌─────────────────────────────────────────────────────────────┐
│  User Input                                                  │
└────────────┬────────────────────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────────────────────┐
│  MCPClient.process_query()                                   │
│  - Add to conversation history                               │
│  - Format for LLM                                            │
└────────────┬────────────────────────────────────────────────┘
             │
             ▼
┌─────────────────────────────────────────────────────────────┐
│  LLM (Gemini/Claude)                                         │
│  - Analyze query with available tools                        │
│  - Decide: respond directly OR call tools                    │
└────────────┬────────────────────────────────────────────────┘
             │
        ┌────┴────┐
        │         │
        ▼         ▼
   Text Response  Tool Call(s)
        │         │
        │         ▼
        │    ┌─────────────────────────────────────┐
        │    │  _execute_tool_call()                │
        │    │  - Call MCP server via stdio         │
        │    └────────────┬────────────────────────┘
        │                 │
        │                 ▼
        │    ┌─────────────────────────────────────┐
        │    │  _extract_tool_result()              │
        │    │  - Parse various formats             │
        │    │  - Normalize to dict                 │
        │    └────────────┬────────────────────────┘
        │                 │
        │                 ▼
        │    ┌─────────────────────────────────────┐
        │    │  Send tool result back to LLM       │
        │    └────────────┬────────────────────────┘
        │                 │
        │                 ▼
        │         [LLM processes result]
        │                 │
        └─────────────────┴──────► Final Response
                                        │
                                        ▼
                              ┌──────────────────────┐
                              │  Display to User     │
                              └──────────────────────┘

# Import necessary libraries
import asyncio  # For handling asynchronous operations
import os       # For environment variable access
import sys      # For system-specific parameters and functions
import json     # For handling JSON data (used when printing function declarations)

# Import MCP client components
from typing import Optional  # For type hinting optional values
from contextlib import AsyncExitStack  # For managing multiple async tasks
from mcp import ClientSession, StdioServerParameters  # MCP session management
from mcp.client.stdio import stdio_client  # MCP client for standard I/O communication

# Import Google's Gen AI SDK
from google import genai
from google.genai import types
from google.genai.types import Tool, FunctionDeclaration
from google.genai.types import GenerateContentConfig
from dotenv import load_dotenv  # For loading API keys from a .env file

# Load environment variables from .env file
load_dotenv()

class MCPClient:
    def __init__(self):
        """Initialize the MCP client and configure the Gemini API."""
        self.session: Optional[ClientSession] = None  # MCP session for communication
        self.exit_stack = AsyncExitStack()  # Manages async resource cleanup
        self.conversation_history = []  # Maintain conversation history

        # Retrieve the Gemini API key from environment variables
        gemini_api_key = os.getenv("GEMINI_API_KEY")
        if not gemini_api_key:
            raise ValueError("GEMINI_API_KEY not found. Please add it to your .env file.")

        # Configure the Gemini AI client
        self.genai_client = genai.Client(api_key=gemini_api_key)

    async def connect_to_server(self, server_script_path: str):
        """Connect to the MCP server and list available tools."""
        # Determine whether the server script is written in Python or JavaScript
        if server_script_path.endswith('.py'):

            # Use uv run to ensure the server runs in its own virtual environment
            server_dir = os.path.dirname(os.path.abspath(server_script_path))
            script_name = os.path.basename(server_script_path)

            # Check if server directory has uv project files
            if os.path.exists(os.path.join(server_dir, 'uv.lock')) or os.path.exists(os.path.join(server_dir, 'pyproject.toml')):
                command = "uv"
                # Use --directory to run from the server's directory
                args = ["run", "--directory", server_dir, "python", script_name]
            else:
                command = "python"
                args = [server_script_path]

        else:
            command = "node"
            args = [server_script_path]

        # Define the parameters for connecting to the MCP server
        server_params = StdioServerParameters(command=command, args=args)

        # Establish communication with the MCP server using standard input/output (stdio)
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))

        # Extract the read/write streams from the transport object
        self.stdio, self.write = stdio_transport

        # Initialize the MCP client session, which allows interaction with the server
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        # Send an initialization request to the MCP server
        await self.session.initialize()

        # Request the list of available tools from the MCP server
        response = await self.session.list_tools()
        tools = response.tools  # Extract the tool list from the response

        # Print a message showing the names of the tools available on the server
        print("\nConnected to server with tools:", [tool.name for tool in tools])

        # Convert MCP tools to Gemini format
        self.function_declarations = convert_mcp_tools_to_gemini(tools)

    def _extract_tool_result(self, result):
        """Extract content from MCP tool result, handling different formats."""
        if not result or not hasattr(result, 'content'):
            return None

        content = result.content
        
        # Handle list of content blocks - collect ALL content, not just first
        if isinstance(content, list) and len(content) > 0:
            extracted_parts = []

            for content_block in content:
                # Check if it's a TextContent object with text attribute
                if hasattr(content_block, 'text'):
                    text = content_block.text
                    # Try to parse as JSON if it looks like JSON
                    try:
                        parsed = json.loads(text)
                        extracted_parts.append(parsed)
                    except (json.JSONDecodeError, TypeError):
                        extracted_parts.append(text)

                # Check if it's a dict with text key
                elif isinstance(content_block, dict):
                    if 'text' in content_block:
                        text = content_block['text']
                        # Try to parse as JSON if it looks like JSON
                        try:
                            parsed = json.loads(text)
                            extracted_parts.append(parsed)
                        except (json.JSONDecodeError, TypeError):
                            extracted_parts.append(text)
                    else:
                        # If it's a dict without 'text', it might be the data itself
                        extracted_parts.append(content_block)

                # Check if it's already a string
                elif isinstance(content_block, str):
                    # Try to parse as JSON
                    try:
                        parsed = json.loads(content_block)
                        extracted_parts.append(parsed)
                    except (json.JSONDecodeError, TypeError):
                        extracted_parts.append(content_block)
                # Try to convert to string
                else:
                    extracted_parts.append(str(content_block))

            # If we have multiple parts, combine them appropriately
            if len(extracted_parts) == 1:
                return extracted_parts[0]
            else:
                # If all parts are lists, flatten them
                if all(isinstance(p, list) for p in extracted_parts):
                    flattened = []
                    for p in extracted_parts:
                        flattened.extend(p)
                    return flattened
                return extracted_parts

        # Handle direct content (string or dict)
        elif isinstance(content, str):
            # Try to parse as JSON
            try:
                return json.loads(content)
            except (json.JSONDecodeError, TypeError):
                return content
        elif isinstance(content, dict):
            return content
        else:
            # Fallback: convert to string
            return str(content) if content else None

    async def _execute_tool_call(self, tool_name: str, tool_args: dict):
        """Execute a tool call and return the result in a format suitable for Gemini."""
        try:
            result = await self.session.call_tool(tool_name, tool_args)
            extracted_result = self._extract_tool_result(result)

            # If the result is a string that looks like JSON, try to parse it
            if isinstance(extracted_result, str):
                try:
                    parsed = json.loads(extracted_result)
                    return {"result": parsed}
                except (json.JSONDecodeError, TypeError):
                    # Not JSON, return as string
                    return {"result": extracted_result}
            
            return {"result": extracted_result}

        except Exception as e:
            print(f"[Error executing tool {tool_name}: {e}]")
            return {"error": str(e), "result": None}

    async def process_query(self, query: str) -> str:
        """
        Process a user query using the Gemini API and execute tool calls if needed.
        Args:
            query (str): The user's input query.
        Returns:
            str: The response generated by the Gemini model.
        """
        # Format user input as a structured Content object for Gemini
        user_prompt_content = types.Content(
            role='user',
            parts=[types.Part.from_text(text=query)]
        )

Content(role='user', parts=[Part(text="What's 2+2?")])

        # Add user message to conversation history
        self.conversation_history.append(user_prompt_content)

        # Send user input to Gemini AI and include available tools for function calling
        response = self.genai_client.models.generate_content(
            model='gemini-2.0-flash-001',
            contents=self.conversation_history,
            config=types.GenerateContentConfig(
                tools=self.function_declarations,
            ),
        )

        # Initialize variables to store final response text
        final_text = []
        max_iterations = 10  # Prevent infinite loops
        iteration = 0

        # Process the response - handle multiple function calls and follow-up responses
        while iteration < max_iterations:
            iteration += 1
            function_calls_made = False

            # Process all candidates in the response
            for candidate in response.candidates:
                if candidate.content.parts:
                    for part in candidate.content.parts:
                        if isinstance(part, types.Part):
                            if part.function_call:

Response
└── Candidates [list]
    └── Candidate
        └── Content
            └── Parts [list]
                ├── Part (text)
                └── Part (function_call)

                                # Extract function call details
                                function_call_part = part
                                tool_name = function_call_part.function_call.name
                                tool_args = function_call_part.function_call.args

                                # Print debug info
                                print(f"\n[Gemini requested tool call: {tool_name} with args {tool_args}]")

[Gemini requested tool call: read_file with args {'path': '/home/user/data.txt'}]

                                # Execute the tool using the MCP server
                                function_response = await self._execute_tool_call(tool_name, tool_args)

                                # Format the tool response for Gemini
                                function_response_part = types.Part.from_function_response(
                                    name=tool_name,
                                    response=function_response
                                )

                                # Structure the tool response as a Content object for Gemini
                                function_response_content = types.Content(
                                    role='tool',
                                    parts=[function_response_part]
                                )

                                # Add function call and response to conversation history
                                self.conversation_history.append(
                                    types.Content(role='model', parts=[function_call_part])
                                )
                                self.conversation_history.append(function_response_content)
                                
                                function_calls_made = True

[user] "What files are in my directory?"
[model] <calls list_files tool>
[tool] ["file1.txt", "file2.py"]
[model] "Your directory contains file1.txt and file2.py"

                            else:
                                # Direct text response (no function call)
                                if hasattr(part, 'text') and part.text:
                                    final_text.append(part.text)

            # If no function calls were made, we're done
            if not function_calls_made:
                break

            # Get follow-up response from Gemini after tool execution
            response = self.genai_client.models.generate_content(
                model='gemini-2.0-flash-001',
                contents=self.conversation_history,
                config=types.GenerateContentConfig(
                    tools=self.function_declarations,
                ),
            )

        # Add final response to conversation history
        if final_text:
            response_text = "\n".join(final_text)
            self.conversation_history.append(
                types.Content(
                    role='model',
                    parts=[types.Part.from_text(text=response_text)]
                )
            )

        # Return the combined response as a single formatted string
        return "\n".join(final_text) if final_text else "No response generated."

    def clear_history(self):
        """Clear the conversation history."""
        self.conversation_history = []

    async def chat_loop(self):
        """Run an interactive chat session with the user."""
        print("\nMCP Client Started! Type 'quit' to exit, 'clear' to clear history.")
        while True:
            query = input("\nQuery: ").strip()
            if query.lower() == 'quit':
                break
            if query.lower() == 'clear':
                self.clear_history()
                print("Conversation history cleared.")
                continue
            
            # Process the user's query and display the response
            response = await self.process_query(query)
            print("\n" + response)

    async def cleanup(self):
        """Clean up resources before exiting."""
        await self.exit_stack.aclose()

def clean_schema(schema):
    """
    Recursively removes 'title' fields from the JSON schema.
    Args:
        schema (dict): The schema dictionary.
    Returns:
        dict: Cleaned schema without 'title' fields.
    """
    if isinstance(schema, dict):
        schema.pop("title", None)  # Remove title if present
        # Recursively clean nested properties
        if "properties" in schema and isinstance(schema["properties"], dict):
            for key in schema["properties"]:
                schema["properties"][key] = clean_schema(schema["properties"][key])
    return schema

# Before
{"type": "object", "title": "MyObject", "properties": {"name": {"type": "string", "title": "Name"}}}

# After
{"type": "object", "properties": {"name": {"type": "string"}}}

def convert_mcp_tools_to_gemini(mcp_tools):
    """
    Converts MCP tool definitions to the correct format for Gemini API function calling.
    Args:
        mcp_tools (list): List of MCP tool objects with 'name', 'description', and 'inputSchema'.
    Returns:
        list: List of Gemini Tool objects with properly formatted function declarations.
    """
    gemini_tools = []
    for tool in mcp_tools:
        # Ensure inputSchema is a valid JSON schema and clean it
        parameters = clean_schema(tool.inputSchema.copy() if hasattr(tool.inputSchema, 'copy') else tool.inputSchema)

        # Construct the function declaration
        function_declaration = FunctionDeclaration(
            name=tool.name,
            description=tool.description,
            parameters=parameters
        )

        # Wrap in a Tool object
        gemini_tool = Tool(function_declarations=[function_declaration])
        gemini_tools.append(gemini_tool)

    return gemini_tools

MCP Tool → FunctionDeclaration → Tool → List of Tools

async def main():
    """Main function to start the MCP client."""
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        # Connect to the MCP server and start the chat loop
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        # Ensure resources are cleaned up
        await client.cleanup()

if __name__ == "__main__":
    # Run the main function within the asyncio event loop
    asyncio.run(main())

model='gemini-2.0-flash-001'

from google import genai

gemini_api_key = os.getenv("GEMINI_API_KEY")
self.genai_client = genai.Client(api_key=gemini_api_key)

types.Content(
    role='user',  # or 'model' or 'tool'
    parts=[
        types.Part.from_text(text="Hello"),
        # Can also include images, function calls, etc.
    ]
)

# Remove Gemini
uv remove google-genai

# Add Claude
uv add anthropic

# Change from:
# GEMINI_API_KEY=your_key_here

# To:
ANTHROPIC_API_KEY=your_key_here

# OLD (Gemini)
from google import genai
from google.genai import types
from google.genai.types import Tool, FunctionDeclaration
from google.genai.types import GenerateContentConfig

# NEW (Claude)
from anthropic import Anthropic

def __init__(self):
    """Initialize the MCP client and configure the Claude API."""
    self.session: Optional[ClientSession] = None
    self.exit_stack = AsyncExitStack()
    self.conversation_history = []  # Format: [{"role": "user", "content": "..."}, ...]
    
    # Retrieve the Anthropic API key
    anthropic_api_key = os.getenv("ANTHROPIC_API_KEY")
    if not anthropic_api_key:
        raise ValueError("ANTHROPIC_API_KEY not found. Please add it to your .env file.")
    
    # Configure the Anthropic client
    self.anthropic_client = Anthropic(api_key=anthropic_api_key)

def convert_mcp_tools_to_claude(mcp_tools):
    """
    Converts MCP tool definitions to Claude's tool format.
    Args:
        mcp_tools (list): List of MCP tool objects.
    Returns:
        list: List of tool definitions in Claude format.
    """
    claude_tools = []
    for tool in mcp_tools:
        parameters = clean_schema(
            tool.inputSchema.copy() if hasattr(tool.inputSchema, 'copy') else tool.inputSchema
        )
        
        claude_tool = {
            "name": tool.name,
            "description": tool.description,
            "input_schema": parameters
        }
        claude_tools.append(claude_tool)
    
    return claude_tools

# OLD (Gemini)
self.function_declarations = convert_mcp_tools_to_gemini(tools)

# NEW (Claude)
self.claude_tools = convert_mcp_tools_to_claude(tools)

async def process_query(self, query: str) -> str:
    """
    Process a user query using Claude API and execute tool calls if needed.
    Args:
        query (str): The user's input query.
    Returns:
        str: The response generated by Claude.
    """
    # Add user message to conversation history
    self.conversation_history.append({
        "role": "user",
        "content": query
    })
    
    max_iterations = 10
    iteration = 0
    
    while iteration < max_iterations:
        iteration += 1
        
        # Call Claude with conversation history and available tools
        response = self.anthropic_client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=4096,
            messages=self.conversation_history,
            tools=self.claude_tools
        )
        
        # Add assistant response to history
        assistant_message = {
            "role": "assistant",
            "content": response.content
        }
        self.conversation_history.append(assistant_message)
        
        # Check if Claude wants to use tools
        tool_uses = [block for block in response.content if block.type == "tool_use"]
        
        if not tool_uses:
            # No tools needed, extract text and return
            text_blocks = [block.text for block in response.content if hasattr(block, 'text')]
            return "\n".join(text_blocks) if text_blocks else "No response generated."
        
        # Execute all tool calls
        tool_results = []
        for tool_use in tool_uses:
            print(f"\n[Claude requested tool call: {tool_use.name} with args {tool_use.input}]")
            
            result = await self._execute_tool_call(tool_use.name, tool_use.input)
            
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": json.dumps(result)
            })
        
        # Add tool results as a user message
        self.conversation_history.append({
            "role": "user",
            "content": tool_results
        })
    
    return "Maximum iterations reached without final response."

uv run client.py <path_to_server_script>

# Python MCP server with uv
uv run client.py ../my-mcp-server/server.py

# Python MCP server (plain)
uv run client.py /path/to/server.py

# Node.js MCP server
uv run client.py ./server.js

Connected to server with tools: ['read_file', 'write_file', 'list_directory']

MCP Client Started! Type 'quit' to exit, 'clear' to clear history.

Query: 

Query: What files are in my current directory?

[Gemini requested tool call: list_directory with args {'path': '.'}]

Your current directory contains:
- client.py
- pyproject.toml
- .env
- README.md

Query: Read the pyproject.toml file

[Gemini requested tool call: read_file with args {'path': 'pyproject.toml'}]

The pyproject.toml file contains your project configuration:
- Project name: mcp-client
- Python version: >=3.13
- Dependencies: google-genai, mcp, python-dotenv

Query: quit

Connected to server with tools: ['tool1', 'tool2', 'tool3']
MCP Client Started! Type 'quit' to exit, 'clear' to clear history.

Error: Could not connect to server

Query: Use the [tool_name] tool to [do something]

[Gemini requested tool call: tool_name with args {...}]
[Tool response shown]

I don't have access to that tool.

Query: My favorite color is blue

Response: Got it, I'll remember that your favorite color is blue.

Query: What's my favorite color?

Response: Your favorite color is blue.

import logging
logging.basicConfig(level=logging.DEBUG)

print(f"[DEBUG] Conversation has {len(self.conversation_history)} messages")

print(f"[DEBUG] Tool returned: {type(extracted_result)} = {extracted_result}")

uv add mcp google-genai python-dotenv

# Test server directly
python path/to/server.py

# Or for Node.js
node path/to/server.js

# Check server has dependencies
cd path/to/server
uv sync  # for Python with uv
# or npm install for Node.js

# Good
uv run client.py ../server/server.py

# Also good
uv run client.py C:/Users/You/server/server.py

# Or in Python code use raw string
r"C:\Users\You\server\server.py"

mcp-client/
├── .venv/                      # Virtual environment (created by uv venv)
│   ├── Scripts/                # (Windows) or bin/ (Unix)
│   ├── Lib/
│   └── ...
├── client.py                   # Main MCP client code (YOU CREATE THIS)
├── main.py                     # Boilerplate (can be ignored or deleted)
├── pyproject.toml              # Project metadata and dependencies
├── uv.lock                     # Locked dependency versions
├── .env                        # API keys (YOU CREATE THIS)
├── .gitignore                  # Git ignore rules
└── README.md                   # This documentation

# In __init__
self.sessions = []  # List of sessions

# Create multiple connections
await client.connect_to_server("server1.py")
await client.connect_to_server("server2.js")

# Merge tool lists
all_tools = self.sessions[0].tools + self.sessions[1].tools

response_stream = self.genai_client.models.generate_content_stream(
    model='gemini-2.0-flash-001',
    contents=self.conversation_history,
    config=types.GenerateContentConfig(tools=self.function_declarations),
)

for chunk in response_stream:
    if chunk.text:
        print(chunk.text, end='', flush=True)

import tkinter as tk
from tkinter import scrolledtext

class MCPClientGUI:
    def __init__(self):
        self.root = tk.Tk()
        self.root.title("MCP Client")
        
        self.chat_display = scrolledtext.ScrolledText(self.root, wrap=tk.WORD)
        self.chat_display.pack(expand=True, fill='both')
        
        self.input_field = tk.Entry(self.root)
        self.input_field.pack(fill='x')
        self.input_field.bind('<Return>', self.send_message)
    
    def send_message(self, event):
        query = self.input_field.get()
        # Process with MCP client...

uv add streamlit

import streamlit as st

st.title("MCP Client")

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("Ask me anything"):
    response = asyncio.run(client.process_query(prompt))
    st.session_state.messages.append({"role": "assistant", "content": response})

async def _execute_tool_call_with_retry(self, tool_name: str, tool_args: dict, max_retries=3):
    """Execute tool call with automatic retry on failure."""
    for attempt in range(max_retries):
        try:
            return await self._execute_tool_call(tool_name, tool_args)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            print(f"[Retry {attempt + 1}/{max_retries}] Tool call failed: {e}")
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

import json
from datetime import datetime

def save_conversation(self, filename=None):
    """Save conversation history to a JSON file."""
    if filename is None:
        filename = f"conversation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    
    with open(filename, 'w') as f:
        json.dump(self.conversation_history, f, indent=2)
    
    print(f"Conversation saved to {filename}")

def load_conversation(self, filename):
    """Load conversation history from a JSON file."""
    with open(filename, 'r') as f:
        self.conversation_history = json.load(f)
    
    print(f"Conversation loaded from {filename}")

# In chat_loop, add new commands:
if query.lower() == 'save':
    self.save_conversation()
    continue
if query.lower().startswith('load '):
    filename = query[5:].strip()
    self.load_conversation(filename)
    continue

def set_model(self, model_name: str):
    """Change the LLM model being used."""
    valid_models = [
        'gemini-2.0-flash-001',
        'gemini-2.0-pro-001',
        'gemini-1.5-flash',
        'gemini-1.5-pro'
    ]
    
    if model_name not in valid_models:
        print(f"Invalid model. Choose from: {valid_models}")
        return
    
    self.current_model = model_name
    print(f"Switched to model: {model_name}")

from collections import Counter

class MCPClient:
    def __init__(self):
        # ... existing code ...
        self.tool_usage_stats = Counter()
    
    async def _execute_tool_call(self, tool_name: str, tool_args: dict):
        self.tool_usage_stats[tool_name] += 1
        # ... rest of existing code ...
    
    def show_stats(self):
        """Display tool usage statistics."""
        print("\n=== Tool Usage Statistics ===")
        for tool, count in self.tool_usage_stats.most_common():
            print(f"{tool}: {count} times")

def __init__(self, system_prompt=None):
    # ... existing code ...
    
    if system_prompt:
        # For Gemini, add as first user message
        self.conversation_history.append(
            types.Content(
                role='user',
                parts=[types.Part.from_text(text=f"System: {system_prompt}")]
            )
        )

client = MCPClient(
    system_prompt="You are a helpful coding assistant. Always explain your reasoning."
)

async def _execute_tool_calls_parallel(self, tool_calls):
    """Execute multiple tool calls concurrently."""
    tasks = [
        self._execute_tool_call(call.name, call.args) 
        for call in tool_calls
    ]
    return await asyncio.gather(*tasks)

uv add fastapi uvicorn

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
client = None  # Initialize globally

class QueryRequest(BaseModel):
    query: str
    session_id: str = "default"

@app.post("/query")
async def process_query_endpoint(request: QueryRequest):
    try:
        response = await client.process_query(request.query)
        return {"response": response, "session_id": request.session_id}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/clear")
async def clear_history_endpoint():
    client.clear_history()
    return {"status": "cleared"}

# Run with: uvicorn your_file:app --reload