> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/BerriAI/litellm/llms.txt
> Use this file to discover all available pages before exploring further.

# Model Context Protocol (MCP)

> Integrate MCP servers and tools into your LiteLLM workflows

## 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

```bash theme={null}
pip install litellm mcp
```

## Quick Start

<Steps>
  <Step title="Create MCP Client">
    Initialize a client for your MCP server:

    ```python theme={null}
    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"
    )
    ```
  </Step>

  <Step title="List Available Tools">
    Discover tools provided by the server:

    ```python theme={null}
    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']}")
    ```
  </Step>

  <Step title="Call a Tool">
    Execute tools with parameters:

    ```python theme={null}
    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)
    ```
  </Step>
</Steps>

## Tool Integration

### Load MCP Tools for LLM Use

Convert MCP tools to OpenAI-compatible format:

```python theme={null}
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:

```python theme={null}
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

<CodeGroup>
  ```python HTTP Transport theme={null}
  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
  )
  ```

  ```python SSE Transport theme={null}
  client = MCPClient(
      server_url="https://mcp-server.example.com/sse",
      transport_type="sse",
      auth_type="api_key",
      auth_value="your-api-key",
      timeout=60.0
  )
  ```

  ```python Stdio Transport theme={null}
  from litellm.types.mcp import MCPStdioConfig

  stdio_config = MCPStdioConfig(
      command="npx",
      args=["-y", "@modelcontextprotocol/server-memory"],
      env={"NODE_ENV": "production"}
  )

  client = MCPClient(
      transport_type="stdio",
      stdio_config=stdio_config
  )
  ```
</CodeGroup>

## Authentication

### Bearer Token

```python theme={null}
client = MCPClient(
    server_url="https://api.example.com",
    auth_type="bearer_token",
    auth_value="your-bearer-token"
)
```

### Basic Auth

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

### API Key

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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:

<Accordion title="MCP to OpenAI Tool Format">
  ```python theme={null}
  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)
  ```
</Accordion>

<Accordion title="OpenAI to MCP Request Format">
  ```python theme={null}
  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}")
  ```
</Accordion>

### SSL Configuration

```python theme={null}
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:

```python theme={null}
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

<Accordion title="Full LLM + MCP Integration">
  ```python theme={null}
  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())
  ```
</Accordion>

## Error Handling

```python theme={null}
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

<Accordion title="Session Management">
  Always use `run_with_session` to ensure proper cleanup:

  ```python theme={null}
  # 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
  ```
</Accordion>

<Accordion title="Tool Validation">
  Validate tool schemas before using with LLMs:

  ```python theme={null}
  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)
  ```
</Accordion>

<Accordion title="Error Recovery">
  Implement graceful degradation:

  ```python theme={null}
  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 []
  ```
</Accordion>

## Reference

### Source Code

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

### Related Documentation

* [Function Calling](/docs/completion/function_call)
* [Custom Tools](/docs/completion/tools)
* [MCP Official Spec](https://spec.modelcontextprotocol.io/)