Quick Adaptation for MCP Protocol Access in Agents

1. Introduction

In LLM application development, a common problem is the need to write adaptation code for each external API, handling various proprietary protocols. MCP (Model Context Protocol) solves this by standardizing the interface between LLMs and external tools, unifying the interaction.

This article covers the principles and practice of setting up an MCP Server, configuring an Agent client, and achieving plug-and-play tool invocation, along with common pitfalls. After reading, you will understand the core mechanisms of the MCP protocol, be able to independently build an MCP Server and integrate it with an Agent, and have the ability to troubleshoot and optimize.

2. Core Concepts of the MCP Protocol

2.1 Protocol Positioning

MCP is a universal interface between LLMs and external tools, based on JSON-RPC 2.0 over HTTP/SSE. It defines two core operations:

  • Tool Discovery (listTools): The Agent requests the list of available tools from the Server. The response includes the tool name, parameter schema (JSON Schema format), and description.
  • Tool Invocation (callTool): The Agent calls a specific tool with parameters, and the Server executes and returns the result.

This standardized design means developers only need to learn how to connect to an MCP Server, without adapting to various proprietary protocols. Regardless of the underlying API format, as long as it is encapsulated as an MCP tool, the Agent can automatically identify and invoke it.

2.2 Roles and Lifecycle

The MCP architecture includes two core roles:

  • MCP Client (Agent side): Interacts with the LLM and sends tool discovery and invocation requests to the MCP Server. Typical implementations include MCP integration modules in frameworks like LangChain and Semantic Kernel.
  • MCP Server (Tool provider): Encapsulates external capabilities (e.g., database queries, API calls, file operations) and exposes them to the Agent via the standard MCP interface. A Server can register multiple tools.

A complete tool invocation flow:

  1. When the Agent starts, it sends a listTools request to the registered MCP Server to obtain the tool list.
  2. The LLM decides to invoke a tool based on the user’s question; the Agent constructs a callTool request and sends it to the corresponding Server.
  3. The Server executes the tool logic and returns the result in JSON format.
  4. The Agent passes the result back to the LLM, which generates the final response.

In practice, it is recommended to cache the tool discovery results to avoid repeated requests before each LLM call. The cache validity period can be set based on the tool update frequency, typically 5-30 minutes.

3. Quick Setup of an MCP Server

3.1 Environment Preparation

Requires Python 3.8+ and installation of the MCP official SDK:

1
pip install mcp

The MCP SDK provides the @mcp.tool() decorator for quickly defining tools, and app.run() to start the service. The SDK supports both stdio and SSE transport methods, which will be explained separately below.

3.2 Minimal Implementation

Below is a complete Server implementation for an echo tool:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from mcp import App, Tool

# Create MCP application instance
app = App("echo-server")

@app.tool()
def echo(message: str) -> str:
    """
    Returns the input message.
    
    Args:
        message: The text content to echo
    """
    return f"Echo: {message}"

if __name__ == "__main__":
    # Run the service, default stdio transport
    app.run(transport="stdio")

Start the service:

1
python echo_server.py

By default, stdio transport is used, suitable for scenarios where communication with the Agent’s subprocess is required. For remote access, switch to SSE:

1
app.run(transport="sse", host="0.0.0.0", port=8000)

3.3 Configuration Points

Tool configuration must strictly follow the MCP protocol specifications to ensure automatic discovery by the Agent:

  • Tool name: Use lowercase letters and underscores, avoid special characters. For example, get_weather not Get Weather.
  • Parameter Schema: Use JSON Schema to define the type, description, and required status of each parameter. For complex structures, it is recommended to use the pydantic model to assist definition.
  • Response format: Always return JSON-compatible data. MCP requires the return value to be a serializable object, including basic types, dictionaries, and lists.

Tool descriptions should clearly state the function, as the LLM uses the description to decide when to call the tool. Descriptions that are too short will reduce trigger accuracy, while too long will increase token consumption. In practice, it is recommended to keep it within 50-100 words.

Error handling must be unified: if the tool fails, return an object containing an error field instead of throwing an exception. For example:

1
{"error": "API request timeout", "code": 500}

After receiving the error, the Agent can retry based on strategy or inform the LLM.

4. Agent-Side MCP Configuration

4.1 Connection Methods

The Agent side needs to use the MCP Client class to connect to the Server. Typical implementations are as follows:

Connection via subprocess (stdio):

1
2
3
4
5
from mcp import MCPClient

# Start MCP Server subprocess, communicate via stdio
client = MCPClient()
client.connect_process(["python", "echo_server.py"])

Connection via HTTP (SSE):

1
2
client = MCPClient()
client.connect_sse("http://localhost:8000/mcp")

Note: The connection method must match the Server’s transport parameter. If the Server uses stdio, the Client must use connect_process; if SSE, use connect_sse; otherwise, communication will fail.

4.2 Tool Discovery and Caching

After a successful connection, call list_tools() to obtain the tool list:

1
2
3
tools = await client.list_tools()
for tool in tools:
    print(f"Tool: {tool.name}, Description: {tool.description}")

In practice, it is recommended to cache the tool list in the Agent context object to avoid re-discovery before each LLM inference. The MCP protocol itself does not provide cache control; you need to implement it yourself.

1
2
3
4
5
6
7
8
9
10
11
class AgentContext:
    def __init__(self, client):
        self.client = client
        self.tools_cache = None
        self.cache_time = 0
    
    async def get_tools(self):
        if not self.is_cache_valid():
            self.tools_cache = await self.client.list_tools()
            self.cache_time = time.time()
        return self.tools_cache

4.3 Invocation Example

Send a request via call_tool(name, arguments):

1
2
result = await client.call_tool("echo", {"message": "Hello MCP"})
print(result)  # {"content": "Echo: Hello MCP"}

Handling errors and timeouts:

1
2
3
4
5
6
7
8
9
try:
    result = await asyncio.wait_for(
        client.call_tool("fetch_data", {"url": url}),
        timeout=10.0
    )
    if "error" in result:
        print(f"Tool execution error: {result['error']}")
except asyncio.TimeoutError:
    print("Tool call timeout")

5. Practical Example: Weather Query Agent Development

5.1 Scenario Description

Build an Agent where the user asks “What’s the weather in Beijing today?” and the Agent calls a weather tool on the MCP Server, returning real-time weather data.

5.2 Server-Side Implementation

Encapsulate an external weather API as an MCP tool:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import requests
from mcp import App

app = App("weather-server")

@app.tool()
def get_weather(city: str) -> dict:
    """
    Get real-time weather for a specified city.
    
    Args:
        city: City name, e.g., "Beijing"
    """
    api_key = "your_api_key"
    url = f"https://api.weather.com/v1/current?city={city}&key={api_key}"
    
    try:
        resp = requests.get(url, timeout=5)
        resp.raise_for_status()
        data = resp.json()
        return {
            "city": city,
            "temperature": data["temp"],
            "condition": data["weather"][0]["description"],
            "humidity": data["humidity"]
        }
    except requests.RequestException as e:
        return {"error": f"Weather API request failed: {str(e)}"}

if __name__ == "__main__":
    app.run(transport="sse", host="0.0.0.0", port=8001)

Note: The timeout should match the external API’s response speed. Public APIs usually respond within 3 seconds; setting 5 seconds avoids long Agent waits. The timeout parameter should set both connection timeout and read timeout.

5.3 Agent-Side Code

Using LangChain as an example, bind MCP tools:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from mcp import MCPClient

# Connect to MCP Server
client = MCPClient()
client.connect_sse("http://localhost:8001/mcp")

# Get tool list and convert to LangChain format
mcp_tools = await client.list_tools()
langchain_tools = [
    Tool(
        name=tool.name,
        func=lambda kwargs: client.call_tool(tool.name, kwargs),
        description=tool.description
    )
    for tool in mcp_tools
]

# Initialize Agent
llm = OpenAI(temperature=0)
agent = initialize_agent(
    langchain_tools,
    llm,
    agent="zero-shot-react-description",
    verbose=True
)

# Execute query
response = agent.run("What's the weather in Beijing today?")
print(response)

When the LLM determines it needs to get weather, it automatically constructs parameters and calls the get_weather tool, returning structured weather data.

6. Plug-and-Play and Modular Expansion

6.1 Dynamic Registration Mechanism

Through a configuration file or registry, the Agent automatically loads all registered MCP Servers at startup, allowing tools to be added/removed without restarting.

Configuration file approach (YAML):

1
2
3
4
5
6
7
8
9
10
servers:
  - name: weather
    transport: sse
    url: http://localhost:8001/mcp
  - name: database
    transport: stdio
    command: ["python", "db_server.py"]
  - name: search
    transport: sse
    url: http://search.internal:8002/mcp

The Agent reads the configuration and batch connects at startup:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import yaml

class MCPAgent:
    def __init__(self, config_path):
        with open(config_path) as f:
            self.server_configs = yaml.safe_load(f)["servers"]
        self.clients = []
    
    async def initialize(self):
        for cfg in self.server_configs:
            client = MCPClient()
            if cfg["transport"] == "sse":
                client.connect_sse(cfg["url"])
            else:
                client.connect_process(cfg["command"])
            self.clients.append(client)
    
    async def discover_all_tools(self):
        all_tools = []
        for client in self.clients:
            tools = await client.list_tools()
            all_tools.extend(tools)
        return all_tools

6.2 Multi-Server Collaboration

When the Agent connects to multiple Servers, the tool lists are automatically merged, and the Agent decides which tool to call based on the LLM’s judgment.

1
2
3
4
5
6
7
agent.add_server("weather", "sse://weather.internal:8001/mcp")
agent.add_server("database", "stdio://python db_server.py")
agent.add_server("search", "sse://search.internal:8002/mcp")

# Agent automatically decides which tool to call
agent.query("Query weather in Shenzhen")
agent.query("Read user info from database")

In traditional proprietary protocol adaptation, integrating each API requires writing separate adaptation code. With MCP, you simply start the corresponding Server, and the Agent calls via a unified interface, significantly improving development efficiency. For example, an API integration that might take 3 days can be completed in half a day with MCP.

6.3 Effect Verification

Comparison between traditional methods and MCP access:

Dimension Traditional Proprietary Protocol Adaptation MCP Access
Adding one tool Write adaptation code, handle authentication, maintain documentation Start MCP Server, configure connection
Integration cycle 2-3 days 2-4 hours
Tool discovery Manually register in Agent configuration Automatic via listTools interface
Version management Each API maintained independently Unified management at Server level

7. Advanced Tips and Pitfall Records

7.1 Pitfall 1: Transport Layer Selection

Issue: Choosing the wrong transport method leads to communication failure.

Analysis:

  • stdio: Suitable for local inter-process communication; data is transmitted via standard input/output streams, low latency, no network overhead. However, the Server must be started in the same process as the Agent.
  • SSE: Suitable for remote services, based on HTTP long connections, supporting cross-network calls. However, it requires handling port occupancy, firewalls, heartbeat keep-alive, etc.

Recommendations:

  • Use stdio for tools on the same host.
  • Use SSE for cross-host or containerized deployments.
  • In SSE scenarios, implement a heartbeat mechanism on the Server side to prevent unexpected disconnection.
1
2
3
4
5
6
7
8
9
10
11
12
# SSE Server heartbeat example
import asyncio
from mcp import App, Transport

class SSEWithHeartbeat(Transport):
    async def run(self):
        async for request in self._listen():
            await self._handle_request(request)
        # Send heartbeat every 30 seconds
        while True:
            await asyncio.sleep(30)
            await self._send_ping()

7.2 Pitfall 2: Authentication and Authorization

Issue: The MCP protocol itself does not provide an authentication mechanism; directly exposing tools poses security risks.

Analysis: If an MCP Server is not configured with authentication, anyone can call the tools, leading to data leakage or operational risks.

Solution: Implement simple Bearer Token validation on the Server side.

1
2
3
4
5
6
7
8
9
10
from mcp import App

app = App("secure-server", auth_token="your_secret_token")

# Agent side carries the Token when connecting
client = MCPClient()
client.connect_sse(
    "http://localhost:8000/mcp",
    headers={"Authorization": "Bearer your_secret_token"}
)

Recommendations: Internal systems can use API Key with IP whitelists; external services should consider OAuth2. Tokens should be rotated regularly and avoid hardcoding in configuration files.

7.3 Pitfall 3: Concurrency and Rate Limiting

Issue: When the Agent concurrently calls multiple tools, a single-threaded Server cannot handle them.

Analysis: The MCP SDK uses synchronous I/O by default; if the Agent calls two tools simultaneously, the second will wait for the first to complete.

Solutions:

  • Use an asynchronous framework (e.g., FastAPI) to run the MCP Server.
  • Set up a connection pool to limit concurrency and avoid resource exhaustion.
1
2
3
4
5
6
7
8
9
from mcp import App
import asyncio

app = App("async-server")

@app.tool()
async def heavy_computation(n: int) -> int:
    await asyncio.sleep(2)  # Simulate time-consuming operation
    return n * 2

Concurrency control example:

1
2
3
4
5
6
7
8
9
10
from concurrent.futures import ThreadPoolExecutor
from mcp import App

app = App("server")
executor = ThreadPoolExecutor(max_workers=5)

@app.tool()
def query_database(sql: str) -> list:
    future = executor.submit(execute_sql, sql)
    return future.result(timeout=30)

7.4 Pitfall 4: Timeout Handling

Issue: The LLM waiting for a tool response times out, causing errors.

Analysis: LLMs typically have their own timeout limits (e.g., 30 seconds). If the tool response exceeds this limit, the LLM will interrupt and report an error.

Recommendations:

  • Set a reasonable timeout threshold on the Client side (recommended 15-20 seconds).
  • Implement a retry mechanism, e.g., automatically retry once on first timeout.
1
2
3
4
5
6
7
8
9
10
11
12
13
class MCPClientWithRetry:
    async def call_tool_with_retry(self, name, arguments, retries=2):
        for attempt in range(retries):
            try:
                result = await asyncio.wait_for(
                    self.client.call_tool(name, arguments),
                    timeout=15.0
                )
                return result
            except asyncio.TimeoutError:
                if attempt == retries - 1:
                    return {"error": "Tool call timed out multiple times"}
                print(f"Timeout, retry attempt {attempt + 2}")
  • For tools that require long execution times, consider switching to an asynchronous task mode: first return a task ID, and the Agent polls for the result.

8. Summary and Extensions

8.1 Core Recap

The MCP protocol simplifies Agent tool invocation development through standardized interfaces, enabling plug-and-play capability expansion. Core benefits include:

  1. Reduced integration complexity: No need to adapt proprietary protocols for each API; just connect to the MCP Server.
  2. Flexible expansion capability: Dynamically register/unregister tools, Agent automatically loads, development cycle shortened by over 60%.
  3. Standardized specification: Based on JSON-RPC 2.0, unified tool discovery and invocation processes, reducing maintenance costs.

8.2 Applicable Scenarios

The MCP protocol is suitable for the following scenarios:

  • Standardization of internal tool chains: Unify internal APIs as MCP tools, allowing Agents to centrally manage invocations.
  • Multi-Agent collaboration: Combined with the A2A protocol, Agents call tools via MCP, then collaborate with other Agents via A2A.
  • Hybrid edge-cloud architecture: Lightweight edge Agents run MCP Servers, cloud Agents remotely invoke via SSE, enabling hybrid inference.

8.3 Future Directions

The community is exploring advanced features:

  • WebSocket transport: Addressing the limitation of SSE’s unidirectional push, supporting bidirectional real-time communication.
  • Mutual authentication and permission levels: Stronger security control based on OAuth2 or mTLS.
  • Tool dependency version management: MCP Server declares dependent tool versions, Agent side automatically checks compatibility.

It is recommended that teams gradually accumulate MCP Server assets in internal projects and establish a unified registry to manage tool endpoints and authentication information. Specific implementation path:

  1. Phase 1: Select 2-3 frequently used external APIs, encapsulate them as MCP Servers, and verify the integration process.
  2. Phase 2: Define internal tool encapsulation standards, requiring new APIs to expose interfaces according to the MCP protocol.
  3. Phase 3: Build a registry to implement automatic tool discovery, health checks, and canary releases.

As the MCP ecosystem matures, it will become the standard protocol for communication between LLM applications and external systems. It is recommended that teams start accumulating practical experience early.

Summary

Through this article, you should have a deeper understanding of the “MCP Protocol Access Agent Tutorial.” It is recommended to practice more with actual projects. If you have any questions, feel free to discuss!