方寸 Portal

AI Engineer Portal

Your personal operating system for career transition.

Private mode
Back to Learning Paths
Lesson

Model Context Protocol (MCP) for tool integration

The emerging standard for connecting AI models to external tools and data sources.

45 min
ai-agents-and-toolsphase-1portfolio

Model Context Protocol (MCP) for tool integration

Why this matters

Every time you wire up a tool to an LLM you are solving the same problem: how does the model know what capabilities exist, and how does your application execute them safely? Until 2024, every team solved this differently — OpenAI function calling, LangChain tools, Anthropic tool use, custom JSON schemas. The result was a Babel of integration patterns that did not compose.

Model Context Protocol (MCP) is Anthropic's open standard that defines a single interface between LLMs and external systems. If a tool is written as an MCP server, any MCP-compatible client (Claude Desktop, Claude API, third-party agents) can use it without modification. Think of it as USB-C for AI tools: one standard connector, many devices.

For application engineers this matters because you can build an MCP server once and reuse it across multiple AI products. It also means the growing ecosystem of pre-built MCP servers (databases, file systems, APIs) works out of the box with your agent.

Core concepts

MCP vs direct function calling vs LangChain tools

Direct function calling is the lowest level: you define a JSON Schema, pass it to the API, and handle the response in your application. It is flexible but tightly coupled — the schema lives in your app code, and you must maintain it there.

LangChain tools wrap functions in a class hierarchy with metadata. They compose well within LangChain but do not work outside it. If you switch providers or remove LangChain, you rewrite your tools.

MCP is a network protocol. An MCP server runs as a separate process (or over HTTP) and exposes capabilities through a standardized API. The client (your application or Claude Desktop) discovers and calls those capabilities. The server knows nothing about the client; the client knows nothing about the server's implementation. This separation is the key architectural advantage.

Direct function callingLangChain toolsMCP
PortabilityApp-specificLangChain-specificAny MCP client
DiscoveryStatic (hardcoded)StaticDynamic (at runtime)
DeploymentIn-processIn-processOut-of-process
EcosystemProvider SDKsLangChain HubMCP registry
ComplexityLowMediumMedium-high

MCP server architecture

An MCP server exposes three types of primitives:

Resources are read-only data sources the model can access. Think of them as context injection: a resource could be a file, a database row, or a live API response. The model can request a resource and get its content back as text.

Tools are callable functions. The model decides to call a tool, the server executes it, and the result comes back. Tools are the MCP equivalent of function calls.

Prompts are reusable prompt templates the server exposes. A server could provide a "summarize document" prompt that the client populates with arguments. This is less commonly used but useful for standardizing prompt patterns across teams.

Transport layers

MCP servers communicate over one of two transports:

  • stdio: The client spawns the server as a subprocess and communicates over stdin/stdout. Simple and secure for local tools.
  • HTTP with SSE: The server runs as a web service. Required for remote tools, shared servers, and multi-user deployments.

Building an MCP server in Python

The mcp Python package provides a high-level server SDK. Install it:

pip install mcp

Here is a minimal MCP server that exposes a database query tool:

# mcp_server.py
import asyncio
import sqlite3
from typing import Any
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, CallToolResult

# Create the server
app = Server("database-query-server")

# Register available tools
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query_database",
            description=(
                "Run a read-only SQL query against the application database. "
                "Use this when the user asks for data that requires filtering, "
                "aggregation, or joining multiple tables. Always use parameterized "
                "queries — never interpolate user input directly into SQL."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "A read-only SQL SELECT statement"
                    },
                    "params": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Positional parameters for the query (? placeholders)",
                        "default": []
                    }
                },
                "required": ["sql"]
            }
        )
    ]

# Implement tool execution
@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    if name != "query_database":
        raise ValueError(f"Unknown tool: {name}")

    sql = arguments["sql"].strip()
    params = arguments.get("params", [])

    # Safety: reject non-SELECT statements
    if not sql.upper().startswith("SELECT"):
        return [TextContent(
            type="text",
            text="Error: only SELECT statements are allowed"
        )]

    try:
        conn = sqlite3.connect("app.db")
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute(sql, params)
        rows = cursor.fetchmany(100)  # Limit results to avoid context overflow
        conn.close()

        if not rows:
            return [TextContent(type="text", text="No rows returned")]

        # Format as a readable table
        headers = list(rows[0].keys())
        lines = [" | ".join(headers)]
        lines.append("-" * len(lines[0]))
        for row in rows:
            lines.append(" | ".join(str(row[h]) for h in headers))

        return [TextContent(type="text", text="\n".join(lines))]

    except sqlite3.Error as e:
        return [TextContent(type="text", text=f"Database error: {e}")]

# Run the server over stdio
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="database-query-server",
                server_version="0.1.0",
            ),
        )

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

MCP client integration with Claude

To use this server from your Python application, use the MCP client alongside the Anthropic SDK:

import asyncio
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_agent_with_mcp():
    # Start the MCP server as a subprocess
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Discover available tools from the server
            tools_result = await session.list_tools()
            mcp_tools = tools_result.tools

            # Convert MCP tool definitions to Anthropic format
            anthropic_tools = [
                {
                    "name": tool.name,
                    "description": tool.description,
                    "input_schema": tool.inputSchema
                }
                for tool in mcp_tools
            ]

            client = anthropic.Anthropic()
            messages = [
                {"role": "user", "content": "How many users signed up in March 2025?"}
            ]

            # Agentic loop
            while True:
                response = client.messages.create(
                    model="claude-opus-4-5",
                    max_tokens=1024,
                    tools=anthropic_tools,
                    messages=messages
                )

                if response.stop_reason == "end_turn":
                    # Extract final text response
                    for block in response.content:
                        if hasattr(block, "text"):
                            print(block.text)
                    break

                # Process tool calls
                messages.append({"role": "assistant", "content": response.content})
                tool_results = []

                for block in response.content:
                    if block.type == "tool_use":
                        # Call the MCP server to execute the tool
                        result = await session.call_tool(
                            block.name,
                            arguments=block.input
                        )
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result.content[0].text if result.content else ""
                        })

                messages.append({"role": "user", "content": tool_results})

asyncio.run(run_agent_with_mcp())

Common mistakes

Registering too many tools at once. MCP servers can expose dozens of tools, but every tool definition consumes context window tokens. In a 200-tool server, the tool schemas alone can consume 15,000+ tokens before the user's message arrives. Scope your MCP server to the tools relevant to the task domain — or implement tool filtering based on context.

Skipping input validation in the server. The MCP client passes tool arguments as-is from the model. The model can hallucinate parameter values. Always validate inputs server-side, even when your JSON schema is correct — schemas tell the model what to send but do not prevent the model from sending garbage.

Running MCP servers without authentication on HTTP transport. Stdio transport is local and inherently sandboxed. HTTP transport is exposed to the network. Add authentication (API keys, OAuth) before deploying an HTTP MCP server, especially if it can access databases or filesystems.

Assuming MCP replaces your function calling code immediately. MCP is newer and not yet universally supported. Direct API function calling still works everywhere. Adopt MCP when you need portability across clients or want to use the growing ecosystem of pre-built servers — not because it is the new thing.

Try it yourself

Extend the database query server above to also expose a list_tables tool that returns all table names in the database and a describe_table tool that returns the column names and types for a given table. Then write a short agent that, given a question like "what is the average order value by country?", first calls list_tables, then describe_table for relevant tables, then constructs and executes a query_database call. The agent should not have any hardcoded knowledge of the database schema.

Lesson Deep Dive

Ask a follow-up question about this lesson and get an AI-powered explanation.

Loading previous questions...