Model Context Protocol (MCP): The Standard for LLM Integration

⏱️ Estimated reading time: 18 minutes

Introduction

The Model Context Protocol (MCP) is an open standard launched in November 2024 that enables seamless, secure integration between Large Language Models (LLMs) and external systems. Developed initially by Anthropic and quickly adopted by major AI vendors including OpenAI, Google DeepMind, and Microsoft, MCP represents a significant step toward standardizing how AI agents interact with tools and data sources.

Why MCP Matters

The Problem It Solves

Before MCP, each LLM provider had proprietary methods for tool integration: - OpenAI's function calling - Anthropic's tool use - Google's extensions - Custom implementations for open-source models

This fragmentation meant: - Developers had to write different integrations for each provider - Tools couldn't be easily shared across platforms - Security and authentication were handled inconsistently - No standard for capability discovery

The MCP Solution

MCP provides: - Universal tool interface that works across all compliant LLMs - Standardized authentication and security protocols - Capability negotiation for discovering available tools - Consistent error handling across implementations

Core Architecture

1. Protocol Layers

┌─────────────────────────────────┐
│         Application             │
├─────────────────────────────────┤
│      MCP Client (LLM)          │
├─────────────────────────────────┤
│    MCP Protocol (JSON-RPC)     │
├─────────────────────────────────┤
│      MCP Server (Tools)        │
├─────────────────────────────────┤
│     External Systems            │
└─────────────────────────────────┘

2. Key Components

MCP Servers

Expose tools and resources to LLMs:

from mcp import Server, Tool, Resource

server = Server("my-tools")

@server.tool()
async def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the database for matching records."""
    # Implementation
    return results

@server.resource()
async def get_file_content(path: str) -> str:
    """Read content from a file."""
    # Implementation
    return content

MCP Clients

LLMs that consume MCP services:

from mcp import Client

client = Client()
await client.connect("mcp://localhost:3000")

# Discover available tools
tools = await client.list_tools()

# Execute a tool
result = await client.call_tool(
    "search_database",
    {"query": "user metrics", "limit": 5}
)

MCP Transport

Communication layer (WebSocket, HTTP, or stdio):

# WebSocket transport
transport = WebSocketTransport("ws://localhost:3000")

# HTTP transport
transport = HTTPTransport("https://api.example.com/mcp")

# Standard I/O transport (for local tools)
transport = StdioTransport()

Protocol Specification

1. Message Format

MCP uses JSON-RPC 2.0 for communication:

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "search_database",
    "arguments": {
      "query": "revenue reports",
      "limit": 10
    }
  },
  "id": "req_123"
}

2. Core Methods

Tool Discovery

// Request
{
  "jsonrpc": "2.0",
  "method": "tools/list",
  "id": "1"
}

// Response
{
  "jsonrpc": "2.0",
  "result": {
    "tools": [
      {
        "name": "search_database",
        "description": "Search the database",
        "inputSchema": {
          "type": "object",
          "properties": {
            "query": {"type": "string"},
            "limit": {"type": "integer", "default": 10}
          },
          "required": ["query"]
        }
      }
    ]
  },
  "id": "1"
}

Tool Execution

// Request
{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "search_database",
    "arguments": {"query": "users", "limit": 5}
  },
  "id": "2"
}

// Response
{
  "jsonrpc": "2.0",
  "result": {
    "content": [
      {"type": "text", "text": "Found 5 users..."},
      {"type": "data", "data": [...]}
    ]
  },
  "id": "2"
}

3. Resource Access

MCP distinguishes between tools (actions) and resources (data):

// List resources
{
  "jsonrpc": "2.0",
  "method": "resources/list",
  "id": "3"
}

// Read resource
{
  "jsonrpc": "2.0",
  "method": "resources/read",
  "params": {
    "uri": "file:///data/config.json"
  },
  "id": "4"
}

Implementation Examples

1. Creating an MCP Server

from mcp import Server, Tool, Resource, Authentication
from mcp.server.stdio import stdio_server
import json
import sqlite3

class DatabaseMCPServer:
    def __init__(self, db_path: str):
        self.server = Server("database-tools")
        self.db_path = db_path
        self.setup_tools()

    def setup_tools(self):
        @self.server.tool()
        async def query_database(
            sql: str,
            params: list = None
        ) -> list[dict]:
            """Execute a SQL query on the database."""
            conn = sqlite3.connect(self.db_path)
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()

            if params:
                cursor.execute(sql, params)
            else:
                cursor.execute(sql)

            results = [dict(row) for row in cursor.fetchall()]
            conn.close()
            return results

        @self.server.tool()
        async def describe_table(table_name: str) -> dict:
            """Get schema information for a table."""
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            # Get column info
            cursor.execute(f"PRAGMA table_info({table_name})")
            columns = cursor.fetchall()

            # Get row count
            cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
            count = cursor.fetchone()[0]

            conn.close()

            return {
                "table": table_name,
                "columns": [
                    {
                        "name": col[1],
                        "type": col[2],
                        "nullable": not col[3],
                        "primary_key": bool(col[5])
                    }
                    for col in columns
                ],
                "row_count": count
            }

        @self.server.resource()
        async def get_schema() -> str:
            """Get the complete database schema."""
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()

            cursor.execute("""
                SELECT sql FROM sqlite_master 
                WHERE type IN ('table', 'index', 'view')
            """)

            schema = "\n".join(row[0] for row in cursor.fetchall())
            conn.close()
            return schema

    async def run(self):
        async with stdio_server(self.server):
            await self.server.wait_for_shutdown()

# Run the server
if __name__ == "__main__":
    import asyncio
    server = DatabaseMCPServer("mydata.db")
    asyncio.run(server.run())

2. Integrating MCP with LangChain

from langchain.tools import BaseTool
from mcp import Client
import asyncio

class MCPTool(BaseTool):
    """Wrapper to use MCP tools in LangChain."""

    def __init__(self, mcp_client: Client, tool_name: str, tool_spec: dict):
        self.client = mcp_client
        self.name = tool_name
        self.description = tool_spec["description"]
        self.args_schema = tool_spec["inputSchema"]

    def _run(self, **kwargs) -> str:
        """Synchronous execution for LangChain."""
        return asyncio.run(self._arun(**kwargs))

    async def _arun(self, **kwargs) -> str:
        """Async execution of MCP tool."""
        result = await self.client.call_tool(self.name, kwargs)
        return result.get("content", "")

# Create LangChain agent with MCP tools
async def create_mcp_agent():
    from langchain.agents import initialize_agent, AgentType
    from langchain_openai import ChatOpenAI

    # Connect to MCP server
    client = Client()
    await client.connect("mcp://localhost:3000")

    # Get available tools
    tool_specs = await client.list_tools()

    # Convert to LangChain tools
    tools = [
        MCPTool(client, spec["name"], spec)
        for spec in tool_specs["tools"]
    ]

    # Create agent
    llm = ChatOpenAI(model="gpt-4")
    agent = initialize_agent(
        tools=tools,
        llm=llm,
        agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION
    )

    return agent

3. MCP with Authentication

from mcp import Server, Authentication, SecurityContext
from mcp.auth import OAuth2Provider, APIKeyProvider

class SecureMCPServer:
    def __init__(self):
        self.server = Server(
            "secure-tools",
            authentication=OAuth2Provider(
                client_id="your-client-id",
                client_secret="your-secret",
                auth_url="https://auth.example.com/oauth/authorize",
                token_url="https://auth.example.com/oauth/token"
            )
        )
        self.setup_tools()

    def setup_tools(self):
        @self.server.tool(require_auth=True)
        async def sensitive_operation(
            ctx: SecurityContext,
            data: str
        ) -> dict:
            """Perform a sensitive operation."""
            # Check user permissions
            if not ctx.has_permission("write"):
                raise PermissionError("Write permission required")

            # Log the operation
            await self.audit_log(
                user=ctx.user_id,
                action="sensitive_operation",
                data=data
            )

            # Perform operation
            return {"status": "success", "user": ctx.user_id}

Security Considerations

1. Authentication Methods

MCP supports multiple authentication mechanisms: - OAuth 2.0: For user-delegated access - API Keys: For service-to-service communication - mTLS: For certificate-based authentication - Custom: Extensible authentication providers

2. Authorization and Permissions

@server.tool(
    permissions=["read:database", "write:logs"],
    rate_limit="100/hour"
)
async def protected_tool(ctx: SecurityContext, data: str) -> dict:
    # Tool only accessible with proper permissions
    pass

3. Data Isolation

MCP servers can implement tenant isolation:

class MultiTenantMCPServer:
    @server.tool()
    async def query_data(ctx: SecurityContext, query: str) -> list:
        tenant_id = ctx.tenant_id
        # Ensure queries are scoped to tenant
        return await self.db.query(query, tenant=tenant_id)

Adoption and Ecosystem

Major Adopters (as of 2025)

Company	Implementation	Status
Anthropic	Claude native support	✅ Production
OpenAI	ChatGPT & API support	✅ Production
Google	Gemini integration	✅ Beta
Microsoft	Azure AI Studio	✅ Production
Meta	Llama tooling	🚧 In Progress
AWS	Bedrock integration	✅ Production

Available MCP Servers

Popular pre-built MCP servers: - mcp-server-sqlite: SQLite database access - mcp-server-filesystem: File system operations - mcp-server-git: Git repository operations - mcp-server-slack: Slack integration - mcp-server-google-drive: Google Drive access - mcp-server-postgres: PostgreSQL database - mcp-server-mongodb: MongoDB operations - mcp-server-elasticsearch: Search capabilities

Best Practices

1. Server Design

• Keep tools focused and single-purpose
• Provide clear, detailed descriptions
• Use structured schemas for inputs/outputs
• Implement proper error handling

2. Security

• Always authenticate clients
• Implement rate limiting
• Validate and sanitize inputs
• Use least-privilege principle
• Audit sensitive operations

3. Performance

• Cache frequently accessed resources
• Implement connection pooling
• Use async operations for I/O
• Batch operations when possible

4. Testing

from mcp.testing import MCPTestClient

async def test_database_tool():
    client = MCPTestClient()
    await client.connect_to_server(DatabaseMCPServer("test.db"))

    # Test tool discovery
    tools = await client.list_tools()
    assert "query_database" in [t["name"] for t in tools["tools"]]

    # Test tool execution
    result = await client.call_tool(
        "query_database",
        {"sql": "SELECT * FROM users LIMIT 1"}
    )
    assert result["content"] is not None

Comparison with Alternatives

Feature	MCP	OpenAI Functions	LangChain Tools	Custom APIs
Standardized	✅	❌ Provider-specific	❌ Framework-specific	❌
Cross-platform	✅	❌	⚠️ Limited	❌
Discovery	✅	⚠️ Manual	⚠️ Manual	❌
Authentication	✅ Built-in	⚠️ Basic	❌	Varies
Type Safety	✅	✅	⚠️	Varies
Streaming	✅	✅	✅	Varies

Future Roadmap

Planned Features (2025)

• Batch Operations: Efficient multi-tool execution
• Caching Protocol: Standardized result caching
• Federation: Cross-server tool discovery
• Versioning: Tool version negotiation
• Observability: Built-in metrics and tracing

Community Contributions

The MCP specification is open source and accepts contributions: - GitHub Repository - Reference Implementations - Community Servers

Getting Started

Installation

# Python SDK
pip install mcp

# TypeScript/JavaScript SDK
npm install @modelcontextprotocol/sdk

# Rust SDK
cargo add mcp

Quick Start Example

# server.py
from mcp.server.stdio import stdio_server
from mcp import Server

server = Server("hello-mcp")

@server.tool()
async def hello(name: str = "World") -> str:
    """Say hello to someone."""
    return f"Hello, {name}!"

if __name__ == "__main__":
    import asyncio
    asyncio.run(stdio_server(server))

# client.py
from mcp import Client
import asyncio

async def main():
    client = Client()
    await client.connect_stdio(["python", "server.py"])

    result = await client.call_tool("hello", {"name": "MCP"})
    print(result["content"])  # "Hello, MCP!"

asyncio.run(main())

Conclusion

The Model Context Protocol represents a crucial step toward standardizing how AI agents interact with external tools and systems. By providing a common interface, robust security, and cross-platform compatibility, MCP enables developers to build tools once and use them everywhere, accelerating the development and deployment of AI agent applications.

Resources

• Official MCP Documentation
• MCP Specification
• Python SDK Documentation
• TypeScript SDK
• Community Forums
• Example Implementations

Next Steps

• Explore Orchestration Frameworks that support MCP
• Learn about Autonomous Agents using MCP tools
• Check out Enterprise Platforms with MCP integration