代码级拆解：MCP资源/提示/工具开发全流程实战

原创

聚客AI

发布于 2025-07-24 17:33:35

41900

代码可运行

文章被收录于专栏：AI大模型应用开发炼丹房AI大模型应用开发炼丹房

运行总次数：0

代码可运行

本文较长，建议点赞收藏，以免遗失。文中实战代码我放在了文末。关注我，实力宠粉。

一、MCP服务架构核心

三层服务模型（资源/提示/工具协同工作）：

服务类型对比矩阵：

服务类型	状态修改	缓存支持	协议类比	典型应用场景
Resource	❌	✅	REST GET	数据查询/配置读取
Prompt	❌	✅	模板引擎	标准化LLM交互
Tool	✅	❌	REST POST/PUT	系统操作/复杂计算

二、开发环境搭建

高效工具链配置：

UV包管理（替代pip/conda）：

# Windows安装
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# 项目初始化
uv init mcp_server
cd mcp_server
uv venv
.\.venv\Scripts\activate

跨平台依赖：

# 核心依赖安装
uv add mcp[cli] httpx psycopg2

Node.js环境（v18.20.8必备）：

在 https://nodejs.org/zh-cn/download，选择 v22.17.0（LTS）版本进行下载并安装：

# 验证安装
node --version
npm --version

三、服务端核心实现

1. 资源(Resource)开发 数据库连接模板：

DB_CONFIG = {
    "dbname": "production_db",
    "user": "admin",
    "password": "secure_pass",
    "host": "10.1.1.27",
    "port": "11003"
}

def get_db_connection():
    return psycopg2.connect(**DB_CONFIG)

四类关键资源：

# 1. 基础测试资源
@mcp.resource("test://hello")
def hello() -> str:
    return "Hello, MCP World!"

# 2. 表名查询
@mcp.resource("db://tables")
def list_tables() -> str:
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT table_name FROM information_schema.tables")
            return json.dumps([row[0] for row in cur.fetchall()])

# 3. 表数据查询（防SQL注入）
@mcp.resource("db://tables/{table_name}/data")
def get_table_data(table_name: str, limit: int = 100) -> str:
    with get_db_connection() as conn:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute("SELECT * FROM %s LIMIT %s", 
                       (psycopg2.extensions.AsIs(table_name), limit))
            return json.dumps(cur.fetchall(), default=str)

# 4. 表结构查询
@mcp.resource("db://tables/{table_name}/schema")
def get_table_schema(table_name: str) -> str:
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT column_name, data_type 
                FROM information_schema.columns 
                WHERE table_name = %s
                """, (table_name,))
            return json.dumps([dict(name=row[0], type=row[1]) for row in cur.fetchall()])

2. 提示(Prompt)开发 省级介绍模板：

@mcp.prompt()
def introduce_province(province: str) -> str:
    return f"""
    请从以下维度介绍{province}：
    1. 历史沿革
    2. 人文地理特点
    3. 经济发展现状
    4. 特色旅游资源
    """

代码调试模板（多轮对话）：

@mcp.prompt()
def debug_code(code: str, error: str) -> list[base.Message]:
    return [
        base.SystemMessage("你是一位专业的代码调试助手"),
        base.UserMessage("请帮我修复以下代码："),
        base.UserMessage(f"```python\n{code}\n```"),
        base.UserMessage(f"错误信息：\n{error}"),
        base.AssistantMessage("我将按以下步骤分析：1. 语法检查 2. 逻辑分析 3. 修复方案")
    ]

3. 工具(Tool)开发 数学运算工具集：

@mcp.tool()
def add(a: float, b: float) -> float:
    """加法运算 (a + b)"""
    return a + b

@mcp.tool()
def divide(a: float, b: float) -> float:
    """除法运算 (a / b)"""
    if b == 0:
        raise ValueError("除数不能为零")
    return a / b

服务启动配置：

if __name__ == "__main__":
    mcp = FastMCP(
        "Production MCP Server",
        debug=False,  # 生产环境关闭调试
        host="0.0.0.0",
        port=8000
    )
    mcp.run('sse')  # 生产环境使用SSE协议

四、MCP Inspector 验证流程

MCP Inspector 是专为 MCP 服务端设计的交互式调试工具，提供了一个直观的界面，使得开发者能够快速地验证服务端的响应和状态。使用 MCP Inspector 来测试验证上述开发的服务端功能。

1. 运行 MCP Inspector

在终端运行 mcp --help，可以查看 mcp 命令的用法（由下面的返回结果可知，可以通过 mcp dev 命令运行 mcp inspector）：

>mcp --help

mcp dev 命令语法：

mcp dev --help

运行命令：mcp dev db_server_see.py（ mcp dev 命令默认使用 stdio 通信方式来启动 mcp 服务端），对上面开发的 mcp 服务端功能进行测试。点击输出的链接，即可打开浏览器，进入 mcp inspector 用户界面：

打开 mcp inspector 的页面，点击左侧的 connect 按钮，连接到 mcp 服务端：

连接到 mcp 服务端后，左侧的页面顶部显示 Resources、Prompts、Tools 三个按钮，可以分别对服务端暴露的 Resources、Prompts、Tools 功能进行测试验证。

2. Resources 功能验证

点击 Resources，然后点击下方的 List Resources、List Templates，可以查看资源 list。

查看数据库的数据表清单，点击资源 URI list_table 后，右侧可以查看到 mcp 服务端返回的内容，目前数据库有两张数据表，分别是 chinese_provinces、chinese_movie_ratings 两张表：

查看具体某张数据表的数据内容。点击 get_table_data 资源，输入表名参数和 limit 参数，点击 Read Resource，可以查看 mcp 服务端返回的数据表数据：

3.Prompts 功能验证

点击 Prompts 下的 List Prompts，列出全部 Prompts：

选择其中一个 Prompt，输入参数（如，广东省），点击 Get Prompt，即可按照预先设定的 prompt 模板，生成 prompt：

4. Tools 功能验证

点击 Tools 下的 List Tools，列出全部工具，选择其中一个，输入参数，点击 Run Tools，即可调用工具，获取返回的运行结果：

通过 mcp inspector，可以快捷高效地测试验证我们开发的 mcp 服务端功能。

五、企业级最佳实践

安全防护措施：

SQL注入防护：

# 错误方式（漏洞）
cur.execute(f"SELECT * FROM {table_name}")

# 正确方式（参数化）
cur.execute("SELECT * FROM %s", (psycopg2.extensions.AsIs(table_name),))

访问控制：

# 生产环境限制访问IP
mcp = FastMCP(host="192.168.1.100")

错误处理：

@mcp.resource("db://secure/data")
def secure_data():
    try:
        # 业务逻辑
    except Exception as e:
        return json.dumps({"error": str(e)})

性能优化方案：

# 1. 连接池管理
from psycopg2.pool import SimpleConnectionPool
pool = SimpleConnectionPool(5, 20, **DB_CONFIG)

# 2. 异步处理
@mcp.resource("db://async")
async def async_data():
    async with pool.getconn() as conn:
        # 异步查询