MCP

一、MCP 概述

MCP 的全称是 Model Context Protocol，可以翻译为“模型上下文协议”。它不是一个具体的 Agent 框架，也不是某个模型厂商专属的插件系统，而是一套面向 AI 应用的开放协议。它的核心目标是：让 AI 应用能够用统一方式连接外部工具、数据源和可复用工作流。

如果没有 MCP，每个 AI 应用都要为不同工具单独写适配逻辑。例如一个 IDE 想连接 GitHub、数据库、本地文件系统和浏览器自动化，就需要分别理解这些服务的 API、鉴权方式、调用格式和返回结构。MCP 把这层集成抽象成 Client 与 Server 的通信协议，让 AI 应用只需要理解 MCP，不同外部能力则通过 MCP Server 暴露出来。

1.1 MCP 是什么

MCP 可以理解为 AI 应用和外部世界之间的“连接层”。它负责描述外部系统能提供什么能力、这些能力如何调用、调用结果如何返回，以及哪些上下文可以安全地交给模型使用。

它和传统 API 调用的区别在于：API 通常面向程序员，MCP 面向“模型应用如何发现和使用能力”。MCP Server 不只是提供接口，还会声明工具名称、工具描述、参数 schema、资源 URI、提示词模板等元信息。Host 可以把这些信息交给模型，让模型判断下一步是否需要调用工具或读取资源。

MCP 典型应用场景包括：

• IDE 读取项目文件、搜索代码、运行测试
• 知识库检索文档、读取笔记、生成摘要
• 数据库查询表结构、执行只读 SQL、分析结果
• 浏览器或桌面工具执行自动化任务
• 将个人脚本封装成模型可调用的工具

1.2 MCP 的整体架构

MCP 架构里有三个核心角色：Host、Client 和 Server。

Host 是用户直接交互的 AI 应用，例如聊天工具、IDE、Agent 平台或桌面助手。Host 负责承载模型交互、用户界面、权限确认和整体工作流。

Client 是 Host 内部的连接器。一个 Host 可以同时创建多个 MCP Client，每个 Client 通常连接一个 MCP Server。Client 负责协议通信、能力发现、工具调用、资源读取和生命周期管理。

Server 是能力提供方。它可以运行在本地，也可以连接远程服务。Server 通过 MCP 暴露 Tools、Resources、Prompts 等能力。例如文件系统 Server 可以暴露“读取文件”“列出目录”“搜索内容”等工具，数据库 Server 可以暴露“查询表结构”“执行 SQL”等工具。

一次典型调用链路如下：

1. 用户在 Host 中提出任务，例如“帮我总结这个项目的 README”。
2. Host 通过 MCP Client 连接文件系统 MCP Server。
3. Client 使用 tools/list 或 resources/list 发现可用能力/资源。
4. 模型根据工具描述或资源描述判断需要读取文件，于是 Host 触发对应工具或资源读取。
5. Server 执行实际操作，把结果返回给 Client。
6. Host 将结果组织进模型上下文，模型生成最终回答。

MCP 底层使用 JSON-RPC 2.0 作为消息格式。JSON-RPC 能表示请求、响应、通知和错误。MCP 在此基础上定义了初始化、能力发现、工具调用、资源读取等具体方法。

1.3 MCP 核心原语

MCP 的学习重点是理解它的“原语”。原语可以理解为协议中最基础、最可组合的能力单元。

Tools

Tools 是模型可以调用的可执行函数，适合表示“会产生动作”的能力。例如读取文件、搜索仓库、调用 API、查询数据库、创建 issue、运行脚本等。每个工具通常包含名称、描述、输入参数 schema 和执行结果。

工具描述非常重要，因为模型会根据描述判断什么时候调用它。如果工具描述模糊，模型可能不用、误用，或者在不该调用的时候调用。

Resources

Resources 是提供上下文的数据源。它们通常通过 URI 标识，可以是文件内容、数据库记录、网页内容、日志片段或其他结构化数据。资源更偏向“读上下文”，工具更偏向“执行动作”。

例如：

file:///project/README.md
postgres://database/schema/users
docs://rag/chunking-notes

Resources 的价值在于让 Host 或模型能够按需读取上下文，而不是把所有数据一次性塞进提示词。

Prompts

Prompts 是可复用的提示词模板或工作流。它适合沉淀常见任务，例如“代码审查模板”“生成测试用例模板”“总结会议记录模板”。Server 暴露 Prompts 后，Host 可以让用户或模型选择并填充这些模板。

Prompts 的意义不是替代模型提示词，而是把团队或个人常用的工作方法协议化、可复用化。

Sampling

Sampling 是一种由 Server 向 Client 请求模型能力的机制。也就是说，Server 在某些场景下可以请求 Host 侧的模型完成一次生成或判断。它适合更复杂的交互式工作流，但也需要更谨慎的权限和用户确认。

Roots

Roots 用来告诉 Server 当前允许访问的边界，例如某些工作目录或 URI 范围。它可以降低 Server 误读、越权读取文件的风险。对于本地文件系统类 Server 来说，Roots 是非常关键的安全边界。

Elicitation

Elicitation 表示 Server 可以向用户请求额外信息。例如某个工具需要确认目标分支、缺少 API 参数，或者需要用户选择一个操作范围时，就可以通过 Elicitation 让 Host 向用户收集信息。

1.4 MCP 通信流程

MCP 的通信流程可以分为初始化、发现、调用和错误处理几个阶段。

连接建立后，Client 和 Server 会先进行初始化。初始化阶段会确认协议版本、能力声明、客户端信息和服务端信息。这个阶段类似“握手”，双方需要先知道彼此支持什么能力，后续才能正常通信。

初始化完成后，Client 会根据需要进行能力发现：

tools/list
resources/list
prompts/list

这些方法不会真正执行工具，而是返回“有哪些能力可以用”。例如 tools/list 会返回工具名称、描述和输入参数 schema。Host 可以把这些信息交给模型，让模型判断下一步是否需要调用工具。

当模型决定执行某个工具时，Client 会发送 tools/call 请求。请求里会包含工具名称和参数。Server 校验参数后执行实际逻辑，并返回执行结果。返回内容应该尽量结构化，让 Host 和模型可以稳定理解。

资源读取则围绕 URI 展开。Client 可以先列出资源，再读取某个资源的内容。资源可以有不同内容类型，例如文本、JSON 或二进制内容。对于学习阶段，可以先重点掌握文本资源和 JSON 资源。

错误处理也很重要。常见错误包括：

• 工具名称不存在
• 参数 schema 校验失败
• 用户没有授权
• 访问路径超出 Roots 范围
• 外部 API 调用失败
• Server 内部异常

一个好的 MCP Server 不应该只返回“失败了”，而要尽量说明失败位置、原因和可恢复建议。

二、MCP Server 开发入门

开发 MCP Server 可以先从最小可用版本开始，不要一上来就设计复杂系统。最小版本只需要暴露一个工具，让客户端可以发现并调用它。

开发步骤可以按下面的顺序推进：

1. 选择 SDK 和运行方式。下面统一使用 Python SDK。
2. 创建 Server 实例，声明基本信息和支持的能力。
3. 定义工具名称、描述和输入参数 schema。
4. 实现工具处理函数，完成实际业务逻辑。
5. 启动 Server，并配置到支持 MCP 的 Host 中。
6. 观察日志，确认初始化、tools/list 和 tools/call 是否正常。

先安装依赖：

pip install "mcp[cli]"

例如一个“读取博客文章标题”的工具，可以用 Python 这样实现：

import os
from pathlib import Path

from mcp.server.fastmcp import FastMCP


mcp = FastMCP("blog-helper")
BLOG_ROOT = Path(os.environ.get("BLOG_ROOT", "D:/MyBlog")).resolve() / "source" / "_posts"


@mcp.tool()
def list_blog_posts(limit: int = 10) -> list[dict[str, str]]:
    """列出博客文章文件名和标题，用于了解当前博客内容结构。"""
    posts: list[dict[str, str]] = []

    for path in sorted(BLOG_ROOT.glob("*.md"), reverse=True)[:limit]:
        title = path.stem
        text = path.read_text(encoding="utf-8")
        for line in text.splitlines():
            if line.startswith("title:"):
                title = line.removeprefix("title:").strip()
                break

        posts.append({"file": path.name, "title": title})

    return posts


if __name__ == "__main__":
    mcp.run(transport="stdio")

这个工具的职责很明确：只列出文章，不负责修改文章、不负责发布、不负责生成摘要。后续如果需要摘要能力，可以再增加 summarize_blog_post 工具，而不是把所有能力都塞进一个工具里。

2.1 工具设计原则

MCP 工具设计的质量，直接影响模型使用工具的稳定性。工具不是越多越好，也不是越“万能”越好。一个工具最好只解决一个清晰的问题。

1. 工具职责要单一。比如“读取文件”和“修改文件”最好拆开，因为它们的风险等级不同，权限控制也不同。如果混在一个工具里，模型可能在只想读取信息时误触发写操作。

2. 参数描述要具体。模型并不会像人类开发者一样阅读完整源码，它主要依赖工具名称、描述和 schema 来判断如何传参。字段名、类型、是否必填、示例含义都要清楚。

3. 返回结果要结构化。能返回 JSON 就尽量返回 JSON，能区分 success、data、error 就不要只返回一整段自然语言。结构化结果更容易被模型继续推理，也更方便 Host 做 UI 展示。

4. 危险操作要有保护。删除文件、提交代码、发邮件、调用付费 API、修改数据库等动作，都应该增加确认、权限控制或只读模式。

5. 失败信息要可诊断。比如参数错误时指出哪个字段不合法，权限错误时说明需要什么授权，路径错误时说明当前允许访问的 Roots 范围。

2.2 安全与权限控制

MCP 把外部能力开放给 AI 应用，因此安全问题必须提前考虑。一个不受限制的 MCP Server 可能让模型访问过多文件、执行危险命令，或者把隐私数据带入上下文。

最基本的原则是最小权限。Server 只应该开放完成任务所需的能力。例如数据库 Server 在多数分析场景下只需要只读查询，不应该默认开放 DROP TABLE、DELETE 或任意写入操作。

本地文件系统类 Server 要特别注意 Roots。Server 应该只在用户授权的目录中运行，避免读取密钥文件、浏览器缓存、系统配置等敏感内容。

输入校验也不能省略。即使工具参数来自模型，也必须按不可信输入处理。常见风险包括命令注入、路径穿越、SQL 注入、非法 URL、超大输入导致资源耗尽等。

对于高风险操作，最好设计显式确认流程。例如工具返回“需要用户确认”，由 Host 展示即将执行的动作、目标路径和影响范围，用户确认后再继续。

审计日志同样有价值。日志可以记录工具调用时间、参数摘要、执行结果和错误信息，方便排查问题，也能帮助复盘模型为什么做出某个操作。

2.3 实战案例

学习 MCP 最好的方式是从自己的工作流里找一个小场景。场景越贴近实际，越能体会 MCP 的价值。

文件系统 MCP Server

文件系统 Server 可以提供读取文件、列出目录、搜索文本等能力。它适合 IDE、写作助手和代码分析工具。学习时建议先做只读能力，再考虑写文件、重命名、删除等高风险操作。

下面是一个只读文件系统 MCP Server。它把可访问范围限制在 MCP_ROOT 环境变量指定的目录中，避免模型读取目录外的文件。

import os
from pathlib import Path

from mcp.server.fastmcp import FastMCP


mcp = FastMCP("readonly-filesystem")
ROOT = Path(os.environ.get("MCP_ROOT", ".")).resolve()


def resolve_safe_path(relative_path: str) -> Path:
    """把用户传入路径限制在 ROOT 内，防止路径穿越。"""
    target = (ROOT / relative_path).resolve()
    if ROOT not in target.parents and target != ROOT:
        raise ValueError(f"path is outside allowed root: {relative_path}")
    return target


@mcp.tool()
def list_files(relative_dir: str = ".", limit: int = 50) -> list[dict[str, str]]:
    """列出指定目录下的文件和子目录。"""
    directory = resolve_safe_path(relative_dir)
    if not directory.is_dir():
        raise ValueError(f"not a directory: {relative_dir}")

    items: list[dict[str, str]] = []
    for path in sorted(directory.iterdir())[:limit]:
        items.append(
            {
                "name": path.name,
                "type": "directory" if path.is_dir() else "file",
                "relative_path": str(path.relative_to(ROOT)).replace("\\", "/"),
            }
        )

    return items


@mcp.tool()
def read_text_file(relative_path: str, max_chars: int = 8000) -> dict[str, object]:
    """读取一个 UTF-8 文本文件。"""
    path = resolve_safe_path(relative_path)
    if not path.is_file():
        raise ValueError(f"not a file: {relative_path}")

    text = path.read_text(encoding="utf-8")
    return {
        "path": str(path.relative_to(ROOT)).replace("\\", "/"),
        "content": text[:max_chars],
        "truncated": len(text) > max_chars,
    }


@mcp.tool()
def search_text(keyword: str, relative_dir: str = ".", limit: int = 20) -> list[dict[str, str]]:
    """在目录内搜索文本关键字。"""
    directory = resolve_safe_path(relative_dir)
    if not directory.is_dir():
        raise ValueError(f"not a directory: {relative_dir}")

    matches: list[dict[str, str]] = []
    for path in directory.rglob("*"):
        if len(matches) >= limit:
            break
        if not path.is_file():
            continue

        try:
            lines = path.read_text(encoding="utf-8").splitlines()
        except UnicodeDecodeError:
            continue

        for line_no, line in enumerate(lines, start=1):
            if keyword in line:
                matches.append(
                    {
                        "path": str(path.relative_to(ROOT)).replace("\\", "/"),
                        "line": str(line_no),
                        "preview": line.strip()[:200],
                    }
                )
                break

    return matches


if __name__ == "__main__":
    mcp.run(transport="stdio")

在 PowerShell 中可以这样限制根目录：

$env:MCP_ROOT = "D:/MyBlog"
python filesystem_server.py

在 bash 中可以写成：

MCP_ROOT="D:/MyBlog" python filesystem_server.py

这个 Server 的重点不是功能多，而是边界清楚：只读、限制根目录、跳过非 UTF-8 文件、返回结构化结果。

数据库 MCP Server

数据库 Server 可以暴露表结构查询、只读 SQL 执行、结果摘要等能力。它适合数据分析、后台排查和报表生成。开发时要优先考虑只读账号、查询超时、结果行数限制和敏感字段脱敏。

下面用 SQLite 实现一个只读数据库 MCP Server。为了降低风险，它只允许执行 SELECT 查询，并且自动追加 LIMIT。

import os
import sqlite3
from pathlib import Path
import re

from mcp.server.fastmcp import FastMCP


mcp = FastMCP("readonly-sqlite")
DB_PATH = Path(os.environ["SQLITE_DB_PATH"]).resolve()
TABLE_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


def connect() -> sqlite3.Connection:
    uri = f"file:{DB_PATH.as_posix()}?mode=ro"
    conn = sqlite3.connect(uri, uri=True)
    conn.row_factory = sqlite3.Row
    return conn


def ensure_select(sql: str) -> str:
    normalized = sql.strip().lower()
    if not normalized.startswith("select"):
        raise ValueError("only SELECT queries are allowed")
    if ";" in normalized[:-1]:
        raise ValueError("multiple statements are not allowed")
    return sql.strip().rstrip(";")


def ensure_table_name(table_name: str) -> str:
    if not TABLE_NAME_RE.match(table_name):
        raise ValueError(f"invalid table name: {table_name}")
    return table_name


@mcp.tool()
def list_tables() -> list[str]:
    """列出数据库中的表。"""
    with connect() as conn:
        rows = conn.execute(
            """
            SELECT name
            FROM sqlite_master
            WHERE type = 'table'
            ORDER BY name
            """
        ).fetchall()

    return [row["name"] for row in rows]


@mcp.tool()
def describe_table(table_name: str) -> list[dict[str, str]]:
    """查看表字段信息。"""
    safe_table_name = ensure_table_name(table_name)
    with connect() as conn:
        rows = conn.execute(f"PRAGMA table_info({safe_table_name})").fetchall()

    return [
        {
            "name": row["name"],
            "type": row["type"],
            "not_null": str(bool(row["notnull"])),
            "primary_key": str(bool(row["pk"])),
        }
        for row in rows
    ]


@mcp.tool()
def query(sql: str, limit: int = 50) -> dict[str, object]:
    """执行只读 SELECT 查询，并限制返回行数。"""
    safe_sql = ensure_select(sql)
    limit = max(1, min(limit, 200))

    if " limit " not in safe_sql.lower():
        safe_sql = f"{safe_sql} LIMIT {limit}"

    with connect() as conn:
        cursor = conn.execute(safe_sql)
        rows = [dict(row) for row in cursor.fetchall()]

    return {
        "sql": safe_sql,
        "row_count": len(rows),
        "rows": rows,
    }


if __name__ == "__main__":
    mcp.run(transport="stdio")

PowerShell 运行方式：

$env:SQLITE_DB_PATH = "D:/data/app.db"
python sqlite_server.py

bash 运行方式：

SQLITE_DB_PATH="D:/data/app.db" python sqlite_server.py

2.4 调试与常见问题

MCP Server 开发时，最常见的问题是客户端连接不上。可以先检查 Server 是否启动、命令路径是否正确、环境变量是否加载、运行目录是否符合预期。

如果连接成功但工具列表为空，通常要检查能力声明和工具注册逻辑。工具没有被注册、schema 格式不正确、初始化阶段返回能力不完整，都可能导致客户端发现不到工具。

如果模型没有按预期调用工具，问题不一定在协议层，也可能是工具描述写得不够清楚。可以检查工具名称是否直观，描述里是否说明了使用场景，参数是否过于复杂。

如果工具调用成功但返回内容不可用，要关注返回格式。返回一大段没有结构的文本，会增加模型理解成本。可以改成结构化 JSON，并提供必要字段，例如 items、summary、warnings、nextActions。

权限问题也很常见。尤其是文件路径、工作目录、API Key、数据库账号和网络访问。调试时建议打印关键路径和配置摘要，但不要把密钥写进日志。

三、MCP Client

MCP Client 是 Host 内部负责连接 MCP Server 的组件。它不直接面对用户，而是承担“协议适配层”的角色：建立连接、初始化会话、发现能力、发送请求、接收响应、处理错误，并把结果交给 Host 和模型使用。

3.1 连接方式

MCP 常见连接方式包括 stdio、Streamable HTTP，以及早期或兼容场景中仍可能见到的 SSE。

stdio

stdio 是本地 MCP Server 最常见的连接方式。Host 启动一个本地进程，通过标准输入和标准输出与 Server 交换 JSON-RPC 消息。

它适合本地工具，例如文件系统、Git、命令行脚本、项目分析工具等。优点是简单、权限边界清晰、部署成本低。缺点是通常只适合本机，不适合多个客户端远程共享同一个 Server。

一个典型配置长这样：

{
  "mcpServers": {
    "blog-helper": {
      "command": "python",
      "args": ["D:/tools/blog-helper/server.py"],
      "env": {
        "BLOG_ROOT": "D:/MyBlog"
      }
    }
  }
}

Streamable HTTP

HTTP 连接适合远程 Server 或需要 Web 部署的场景。Client 通过 HTTP 请求与 Server 通信，Server 可以部署在内网、云服务或企业平台中。

HTTP 的优势是更适合服务化、鉴权、负载均衡和跨设备访问。缺点是实现和安全配置更复杂，需要考虑认证、跨域、网络超时、并发、日志和访问控制。

一个简化的连接配置可以写成：

{
  "mcpServers": {
    "remote-docs": {
      "url": "https://mcp.example.com/mcp",
      "headers": {
        "Authorization": "Bearer ${MCP_TOKEN}"
      }
    }
  }
}

SSE

SSE 是 Server-Sent Events 的缩写，早期 MCP 远程连接中经常被提到。它通过服务端事件流向客户端推送消息，再配合 HTTP 请求完成双向交互。

如果是新项目，更应该优先了解当前官方推荐的 HTTP 传输方式；如果维护旧项目或阅读旧教程，看到 SSE 时要知道它通常是在解决“远程连接和事件流”问题。

3.2 示例

下面用 Python 理解 Client 的主要流程。不同 SDK 版本的导入路径可能会有细微变化，但核心步骤基本一致：创建 transport、连接 Server、发现能力、调用能力。

stdio Client 示例

import asyncio
import os

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


async def main() -> None:
    server_params = StdioServerParameters(
        command="python",
        args=["D:/tools/blog-helper/server.py"],
        env={
            **os.environ,
            "BLOG_ROOT": "D:/MyBlog",
        },
    )

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

            tools = await session.list_tools()
            result = await session.call_tool(
                "list_blog_posts",
                arguments={"limit": 10},
            )

            print(tools)
            print(result)


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

这个示例体现了 Client 的三个动作：连接 Server、发现工具、调用工具。

HTTP Client 示例

import asyncio
import os

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client


async def main() -> None:
    headers = {
        "Authorization": f"Bearer {os.environ['MCP_TOKEN']}",
    }

    async with streamablehttp_client(
        "https://mcp.example.com/mcp",
        headers=headers,
    ) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            resources = await session.list_resources()
            print(resources)


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

HTTP Client 更适合远程文档、企业知识库、内部平台工具等场景。它的重点不只是协议调用，还包括认证、网络可靠性和服务端权限控制。

SSE Client 示例

import asyncio

from mcp import ClientSession
from mcp.client.sse import sse_client


async def main() -> None:
    async with sse_client("https://mcp.example.com/sse") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            prompts = await session.list_prompts()
            print(prompts)


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

SSE 示例主要用于理解旧版或兼容实现。学习时可以知道它的存在，但实践上优先把精力放在 stdio 和 Streamable HTTP 上。

四、小结

MCP 的价值不在于让模型“多一个插件”，而在于把 AI 应用和外部世界之间的连接方式标准化。Host 负责用户体验和模型交互，Client 负责协议连接，Server 负责暴露具体能力。Tools、Resources、Prompts 则是最常用的能力抽象。

学习 MCP 时，可以先用一句话记住它：

MCP 是让模型应用以统一协议发现、读取和调用外部能力的连接层。

后续真正要掌握它，还是要写一个小 Server，把自己的文件、文章或脚本接进来。只有跑通过完整链路，才能理解协议设计背后的工程取舍。

参考资料

• Model Context Protocol 官方文档
• MCP Specification
• JSON-RPC 2.0 Specification