MCP 协议：打造超级 AI 大脑

当前 AI 系统面临的核心挑战是数据接入的碎片化问题。企业和开发者在对接不同数据源时，往往需要为每个系统单独开发定制化的对接方案。这种重复性工作开发成本太高了，而且系统间难以实现真正的互联互通。

Claude 的 MCP 协议正就是为了解决这一行业痛点而生的，它是是由 Anthropic 开发的、针对人工智能 (AI) 模型的一种通信协议。MCP 代表“Model Context Protocol”(模型上下文协议)。该协议的设计旨在增强 AI 系统的消息传递能力，促进不同系统之间的协调与合作。

核心理念：

规范化流程
工程化 SDK 代码
高拓展性和开放理念

架构

MCP 协议由一个 client 和多个 server 组成。用户发出请求后，client 会去调用 server 执行逻辑并返回给用户。

client 就是用户的对话框，server 是一些我们提前写好的代码，用来完成我们想实现的系统调用。client 和 server 都可以自行开发。

核心概念

工具 (Tools)

工具是服务器编写，可由客户端调用并由 LLM 用来执行操作的可执行函数。

当前对其支持较多，主要的逻辑都通过工具来实现。

提示 (Prompts)

可以理解为预设好的提示词模版，用于补充用户的输入。

假设用户输入的是：

def add(a, b):

    return a + b

生成的提示可能是：

Explain how this Python code works:

  def add(a, b):

    return a + b

当前对其的支持相对较少。

资源 (Resources)

提供服务端的资源给客户端。

文件内容
数据库记录
API 响应
实时系统数据
截图和图片
日志文件
更多内容

当前对其支持相对较少，一般只有客户端显示调用的时候才会使用。

工作流程

MCP 的工作流程可以在这里找到：https://github.com/modelcontextprotocol/python-sdk/tree/main/examples/clients/simple-chatbot

简单总结如下：

初始化：客户端访问服务端的接口，获取 tools，prompts，resources 等资源。
客户端接收用户输入，并携带着上述资源的上下文一起发送给大模型
客户端判断通过大模型输出判断是否需要调用资源，如果需要就去调用，并把调用的结果返回给大模型
大模型分析，整合结果后返回给客户端。

提示词

官方 demo 中的提示词如下

system_message = (

    "You are a helpful assistant with access to these tools:\n\n"

    f"{tools_description}\n"

    "Choose the appropriate tool based on the user's question. "

    "If no tool is needed, reply directly.\n\n"

    "IMPORTANT: When you need to use a tool, you must ONLY respond with "

    "the exact JSON object format below, nothing else:\n"

    "{\n"

    '    "tool": "tool-name",\n'

    '    "arguments": {\n'

    '        "argument-name": "value"\n'

    "    }\n"

    "}\n\n"

    "After receiving a tool's response:\n"

    "1. Transform the raw data into a natural, conversational response\n"

    "2. Keep responses concise but informative\n"

    "3. Focus on the most relevant information\n"

    "4. Use appropriate context from the user's question\n"

    "5. Avoid simply repeating the raw data\n\n"

    "Please use only the tools that are explicitly defined above."

)

使用 MCP，我们需要：

模型
MCP Client
MCP Server

下面使用 Claude模型，使用 Claude Desktop 作为 MCP Client，自己开发 MCP Server 进行使用。(也可以直接使用官方的 MCP Server)

Server 开发流程

使用 python sdk 开发，前置需要 Mac，并安装 python uv 和 Claudedesktop。

官网给出的开发流程：https://modelcontextprotocol.io/quickstart/server

创建环境

使用 uv 创建开发环境

# Create a new directory for our project

uv init 项目名

cd 项目名



# Create virtual environment and activate it

uv venv

source .venv/bin/activate



# Install dependencies

uv add "mcp[cli]" httpx

编写 mcp server 代码

官网上有 demo (https://modelcontextprotocol.io/quickstart/server) 可以直接丢给大模型仿照这个写。

写提示词的时候要注意下 mcp server 的提示词来源，阅读源代码发现它有两种方式：用户手动指定 description，就是自己写提示词，如果用户不手动指定的话，就默认使用 mcp server 代码的注释作为提示词。

配置 claude desktop

编写完代码，测试能成功运行之后，需要修改 claude desktop 的配置使其识别到这个 server，配置路径：

~/Library/Application\ Support/Claude/claude_desktop_config.json

配置的格式

{

    "mcpServers": {

        "weather": {        

            "command": "uv",   

            "args": [          

                "--directory",   

                "/Users/wpy/downloads/mcp-test",

                "run",

                "hello.py"

            ]

        }

    }

}

保存并退出之后，重启 Claudedesktop，就能看到对话框的右下角出现一个锤子标志：

出现这个标志就表明 desktop 已经识别到了编写的 server，接下来就可以开始使用了。

针对提出的问题，模型会自主判断是否要使用这些 server，并在使用的时候发出提示：

Server demo1：查看本地文件

实现

MCP Server 运行在本地，直接访问本地的文件。

server 代码：

import os

from mcp.server.fastmcp import FastMCP



# Initialize FastMCP server

mcp = FastMCP("file")



@mcp.tool()

async def list_directory_contents(directory: str) -> str:

    """List the contents of a directory.



    Args:

        directory: The path of the directory to list.

    """

    try:

        # Get the list of files and directories in the specified path

        directory_contents = os.listdir(directory)



        if not directory_contents:

            return "The directory is empty."



        # Format the contents into a readable list

        contents = "\n".join(directory_contents)

        return f"Contents of {directory}:\n{contents}"

    except FileNotFoundError:

        return f"Directory {directory} not found."

    except PermissionError:

        return f"Permission denied to access {directory}."

    except Exception as e:

        return f"An error occurred: {str(e)}"





if __name__ == "__main__":

    # Initialize and run the server

    mcp.run(transport='stdio')

Claude desktop 的配置文件：

{

        "file": {

            "command": "uv",

            "args": [

                "--directory",

                "/Users/wpy/downloads/file-mcp",

                "run",

                "hello.py"

            ]

        }

    }

}

效果

可以查看本地目录下的文件。

Server demo2：查询 vlogs 日志

实现

本地运行 vlogs server，通过 MCP 的 Server 去访问。

server 代码：

import os

import requests

from mcp.server.fastmcp import FastMCP



# Initialize FastMCP server

mcp = FastMCP("file")



# Define the URL for the API endpoint

API_URL = "http://localhost:8428/queryLogsByParams"

HEADERS = {

    "Content-Type": "application/json",

    "Authorization": "Bearer my-token"

}





@mcp.tool()

async def get_logs(time: str, namespace: str, app: str, limit: str, pod: list, container: list, keyword: str) -> str:

    """Get logs from the API by passing parameters.



    Args:

        time: The time filter for logs (e.g., "10h").

        namespace: The namespace of the application (e.g., "gpu-operator").

        app: The name of the application (e.g., "gpu-operator").

        limit: The limit for the number of logs to retrieve.

        pod: The list of pods to query.

        container: The list of containers to query.

        keyword: The keyword to filter logs by.

    """

    data = {

        "time": time,

        "namespace": namespace,

        "app": app,

        "limit": limit,

        "jsonMode": "false",

        "stderrMode": "true",

        "numberMode": "false",

        "numberLevel": "h",

        "pod": pod,

        "container": container,

        "keyword": keyword,

        "jsonQuery": []

    }



    try:

        # Make the POST request to the API

        response = requests.post(API_URL, headers=HEADERS, json=data)



        if response.status_code == 200:

            # If the request was successful, return the log data

            return f"Logs retrieved successfully:\n{response.text}"

        else:

            return f"Failed to retrieve logs. Status code: {response.status_code}"



    except requests.exceptions.RequestException as e:

        return f"An error occurred while accessing the API: {str(e)}"





if __name__ == "__main__":

    # Initialize and run the server

    mcp.run(transport='stdio')

Claude desktop 的配置文件：

{

    "mcpServers": {

        "vlogs-server": {

            "command": "uv",

            "args": [

                "--directory",

                "/Users/wpy/downloads/vlogs-mcp",

                "run",

                "hello.py"

            ]

        }

    }

}