Skip to main content

​
Overview

The Model Context Protocol (MCP) standardizes how applications provide context to LLMs. LiteLLM provides a client implementation for connecting to MCP servers and using their tools, prompts, and resources.

​
What is MCP?

MCP enables:
  • Tools: External functions that LLMs can call (e.g., database queries, API calls)
  • Prompts: Pre-configured prompt templates with variables
  • Resources: Context data like files, database records, or API responses

​
Supported Transports

  • HTTP: Connect to HTTP-based MCP servers
  • SSE: Server-Sent Events for real-time updates
  • Stdio: Process-based communication

​
Installation

pip install litellm mcp

​
Quick Start

1

Create MCP Client

Initialize a client for your MCP server:
from litellm.experimental_mcp_client import MCPClient

# HTTP transport
client = MCPClient(
    server_url="http://localhost:8080",
    transport_type="http",
    auth_type="bearer_token",
    auth_value="your-token"
)
2

List Available Tools

Discover tools provided by the server:
from litellm.experimental_mcp_client import load_mcp_tools

async def list_tools():
    async def operation(session):
        return await load_mcp_tools(session, format="openai")
    
    tools = await client.run_with_session(operation)
    
    for tool in tools:
        print(f"Tool: {tool['function']['name']}")
        print(f"Description: {tool['function']['description']}")
3

Call a Tool

Execute tools with parameters:
from mcp.types import CallToolRequestParams

async def call_tool():
    async def operation(session):
        from litellm.experimental_mcp_client import call_mcp_tool
        
        request = CallToolRequestParams(
            name="search_database",
            arguments={"query": "SELECT * FROM users"}
        )
        
        return await call_mcp_tool(session, request)
    
    result = await client.run_with_session(operation)
    print(result)

​
Tool Integration

​
Load MCP Tools for LLM Use

Convert MCP tools to OpenAI-compatible format: 
from litellm.experimental_mcp_client import MCPClient, load_mcp_tools
import litellm

# Initialize MCP client
mcp_client = MCPClient(
    server_url="http://localhost:8080",
    transport_type="http"
)

async def get_tools_for_llm():
    async def operation(session):
        # Get tools in OpenAI format
        return await load_mcp_tools(session, format="openai")
    
    tools = await mcp_client.run_with_session(operation)
    return tools

# Use with LiteLLM
async def main():
    tools = await get_tools_for_llm()
    
    response = await litellm.acompletion(
        model="gpt-4",
        messages=[{"role": "user", "content": "Search for active users"}],
        tools=tools
    )
    
    print(response.choices[0].message)

​
Execute Tool Calls

Handle tool calls from LLM responses: 
from litellm.experimental_mcp_client import call_openai_tool

async def handle_tool_calls(response, mcp_client):
    message = response.choices[0].message
    
    if message.tool_calls:
        for tool_call in message.tool_calls:
            # Execute via MCP
            async def operation(session):
                return await call_openai_tool(session, tool_call)
            
            result = await mcp_client.run_with_session(operation)
            print(f"Tool: {tool_call.function.name}")
            print(f"Result: {result}")

​
Transport Types

from litellm.experimental_mcp_client import MCPClient

client = MCPClient(
    server_url="https://mcp-server.example.com",
    transport_type="http",
    auth_type="bearer_token",
    auth_value="your-api-key",
    timeout=30.0
)

​
Authentication

​
Bearer Token

client = MCPClient(
    server_url="https://api.example.com",
    auth_type="bearer_token",
    auth_value="your-bearer-token"
)

​
Basic Auth

# Provide username:password format
client = MCPClient(
    server_url="https://api.example.com",
    auth_type="basic",
    auth_value="username:password"  # Automatically base64 encoded
)

​
API Key

client = MCPClient(
    server_url="https://api.example.com",
    auth_type="api_key",
    auth_value="your-api-key"  # Sent as X-API-Key header
)

​
Custom Headers

# For custom authentication schemes
client = MCPClient(
    server_url="https://api.example.com",
    auth_type="authorization",
    auth_value="Custom scheme-value"
)

# Or use dictionary for multiple headers
client = MCPClient(
    server_url="https://api.example.com",
    auth_value={
        "X-API-Key": "key123",
        "X-Custom-Header": "value"
    }
)

​
Working with Prompts

​
List Available Prompts

async def list_prompts(client):
    async def operation(session):
        return await session.list_prompts()
    
    prompts = await client.run_with_session(operation)
    
    for prompt in prompts.prompts:
        print(f"Name: {prompt.name}")
        print(f"Description: {prompt.description}")
        print(f"Arguments: {prompt.arguments}")

​
Get a Prompt

from mcp.types import GetPromptRequestParams

async def get_prompt(client, prompt_name, variables):
    async def operation(session):
        request = GetPromptRequestParams(
            name=prompt_name,
            arguments=variables
        )
        return await session.get_prompt(
            name=request.name,
            arguments=request.arguments
        )
    
    result = await client.run_with_session(operation)
    return result

# Use the prompt
prompt_result = await get_prompt(
    client,
    "code_review",
    {"language": "python", "file": "main.py"}
)

print(prompt_result.description)
for message in prompt_result.messages:
    print(message)

​
Working with Resources

​
List Resources

async def list_resources(client):
    resources = await client.list_resources()
    
    for resource in resources:
        print(f"URI: {resource.uri}")
        print(f"Name: {resource.name}")
        print(f"Type: {resource.mimeType}")

​
Read a Resource

from pydantic import AnyUrl

async def read_resource(client, uri: str):
    result = await client.read_resource(AnyUrl(uri))
    
    for content in result.contents:
        if content.type == "text":
            print(content.text)
        elif content.type == "blob":
            print(f"Binary data: {len(content.blob)} bytes")

# Example
await read_resource(client, "file:///path/to/document.txt")

​
Resource Templates

async def list_templates(client):
    templates = await client.list_resource_templates()
    
    for template in templates:
        print(f"URI Template: {template.uriTemplate}")
        print(f"Name: {template.name}")
        print(f"Description: {template.description}")

# Use a template
# Template: "file:///{path}"
await read_resource(client, "file:///documents/report.pdf")

​
Advanced Features

​
Tool Transformation

Manually transform between MCP and OpenAI formats: 
from litellm.experimental_mcp_client.tools import (
    transform_mcp_tool_to_openai_tool
)
from mcp.types import Tool as MCPTool

mcp_tool = MCPTool(
    name="search",
    description="Search the database",
    inputSchema={
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "limit": {"type": "integer"}
        },
        "required": ["query"]
    }
)

openai_tool = transform_mcp_tool_to_openai_tool(mcp_tool)
print(openai_tool)

from litellm.experimental_mcp_client.tools import (
    transform_openai_tool_call_request_to_mcp_tool_call_request
)

# From LLM response
tool_call = response.choices[0].message.tool_calls[0]

# Convert to MCP format
mcp_request = transform_openai_tool_call_request_to_mcp_tool_call_request(
    openai_tool=tool_call
)

print(f"Tool: {mcp_request.name}")
print(f"Arguments: {mcp_request.arguments}")

​
SSL Configuration

import ssl

# Custom SSL context
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

client = MCPClient(
    server_url="https://localhost:8443",
    ssl_verify=ssl_context
)

# Or use CA bundle path
client = MCPClient(
    server_url="https://api.example.com",
    ssl_verify="/path/to/ca-bundle.crt"
)

# Disable SSL verification (not recommended)
client = MCPClient(
    server_url="https://api.example.com",
    ssl_verify=False
)

​
Progress Callbacks

Monitor long-running tool executions: 
async def my_progress_callback(progress: float, total: float | None):
    percentage = (progress / total * 100) if total else 0
    print(f"Progress: {percentage:.1f}%")

async def call_with_progress(client):
    request = CallToolRequestParams(
        name="long_running_task",
        arguments={"data": "large_dataset"}
    )
    
    result = await client.call_tool(
        call_tool_request_params=request,
        host_progress_callback=my_progress_callback
    )
    
    return result

​
Complete Example

import asyncio
from litellm.experimental_mcp_client import (
    MCPClient,
    load_mcp_tools,
    call_openai_tool
)
import litellm

async def main():
    # Initialize MCP client
    mcp_client = MCPClient(
        server_url="http://localhost:8080",
        transport_type="http",
        auth_type="api_key",
        auth_value="your-api-key"
    )
    
    # Load available tools
    async def get_tools(session):
        return await load_mcp_tools(session, format="openai")
    
    tools = await mcp_client.run_with_session(get_tools)
    print(f"Loaded {len(tools)} tools")
    
    # Make LLM request with tools
    response = await litellm.acompletion(
        model="gpt-4",
        messages=[
            {"role": "user", "content": "Find all active users and export to CSV"}
        ],
        tools=tools
    )
    
    message = response.choices[0].message
    
    # Execute tool calls
    if message.tool_calls:
        tool_results = []
        
        for tool_call in message.tool_calls:
            print(f"Executing: {tool_call.function.name}")
            
            async def execute(session):
                return await call_openai_tool(session, tool_call)
            
            result = await mcp_client.run_with_session(execute)
            tool_results.append(result)
        
        # Continue conversation with results
        messages = [
            {"role": "user", "content": "Find all active users and export to CSV"},
            message,
            *[
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": str(result)
                }
                for tc, result in zip(message.tool_calls, tool_results)
            ]
        ]
        
        final_response = await litellm.acompletion(
            model="gpt-4",
            messages=messages
        )
        
        print(final_response.choices[0].message.content)

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

​
Error Handling

import asyncio
from litellm.experimental_mcp_client import MCPClient

async def robust_mcp_call():
    client = MCPClient(
        server_url="http://localhost:8080",
        timeout=30.0
    )
    
    try:
        tools = await client.list_tools()
        print(f"Found {len(tools)} tools")
    except asyncio.CancelledError:
        print("Operation cancelled")
        raise
    except asyncio.TimeoutError:
        print("MCP server timeout")
        return []
    except Exception as e:
        print(f"MCP error: {type(e).__name__}: {e}")
        # Returns empty list on error for graceful degradation
        return []

​
Best Practices

Always use run_with_session to ensure proper cleanup:
# Good: Automatic cleanup
async def operation(session):
    return await session.list_tools()

result = await client.run_with_session(operation)

# Bad: Manual session management (error-prone)
# Don't do this unless you have a specific reason
Validate tool schemas before using with LLMs:
async def get_validated_tools(client):
    async def operation(session):
        tools = await load_mcp_tools(session, format="openai")
        
        # Validate each tool has required fields
        validated = []
        for tool in tools:
            func = tool.get("function", {})
            if func.get("name") and func.get("parameters"):
                validated.append(tool)
        
        return validated
    
    return await client.run_with_session(operation)
Implement graceful degradation:
async def get_tools_with_fallback(client):
    try:
        async def operation(session):
            return await load_mcp_tools(session, format="openai")
        
        return await client.run_with_session(operation)
    except Exception as e:
        print(f"MCP unavailable: {e}")
        # Return empty list or fallback tools
        return []

​
Reference

​
Source Code

  • MCP Client: litellm/experimental_mcp_client/client.py:53
  • Tool utilities: litellm/experimental_mcp_client/tools.py:18

Build docs developers (and LLMs) love

Get started for freeTalk to us