MCP协议实战：让大模型安全连接数据库、API与本地文件

1. MCP 核心概念与架构

客服主管想做一个工单 Agent：用户问「订单 A123 到哪了」，Agent 要查订单库，同时读本地退换货政策。直接给模型一个数据库密码加一个文件目录，风险很大：模型可能误删数据、越权读 /etc/passwd，或者被工单正文里的指令骗着调用写操作。

MCP（Model Context Protocol）把这件事拆成三层：Host 是跑 Agent 的应用，Client 是 Host 内负责连接某个 Server 的连接器，Server 是暴露具体能力的进程。一个 Host 可以挂多个 Client，每个 Client 只连一个 Server。数据库、API、本地文件各自跑独立 Server，权限边界落在进程、账号和网络策略上。

MCP Server 常暴露三类原语：

截至 2026 年，常见规范基线是 2025-06-18。实际开发前先看 MCP 官方规范，SDK 和框架适配可能快于文档。

2. Server 与 Client 最小实现

先写一个只读文件 Server。目标是把路径锁死，功能少一点没关系。下面的 Python 代码用官方 Python SDK 的 FastMCP，只允许读 /srv/mcp-data 下的文本文件，并限制单次读取大小。

from pathlib import Path
from mcp.server.fastmcp import FastMCP
BASE = Path('/srv/mcp-data').resolve()
mcp = FastMCP('file-reader')
@mcp.tool()
def read_text(rel_path: str, max_bytes: int = 65536) -> str:
    p = (BASE / rel_path).resolve()
    if BASE != p and BASE not in p.parents:
        raise ValueError('path outside base')
    if not p.is_file():
        raise ValueError('not a file')
    data = p.read_bytes()[:max_bytes]
    return data.decode('utf-8', errors='replace')
if __name__ == '__main__':
    mcp.run()

resolve() 会把 .. 和符号链接展开，再做父目录检查；漏掉这一步，../../etc/passwd 就能绕过前缀判断。max_bytes 防止一次读进巨大文件把上下文撑爆。

数据库工具用只读连接。SQLite 可以用 URI 的 mode=ro；PostgreSQL 更推荐单独建只读账号，并设置 statement_timeout。

import sqlite3
from mcp.server.fastmcp import FastMCP
mcp = FastMCP('orders-ro')
DB = '/srv/data/orders.db'
@mcp.tool()
def get_order_status(order_id: str) -> dict:
    conn = sqlite3.connect(f'file:{DB}?mode=ro', uri=True)
    try:
        cur = conn.execute(
            'SELECT id,status,updated_at FROM orders WHERE id = ?',
            (order_id,),
        )
        row = cur.fetchone()
        return {'id': row[0], 'status': row[1], 'updated_at': row[2]} if row else {}
    finally:
        conn.close()

写 Client 时先初始化会话，再列工具、调工具。下面是最小 stdio Client：

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
    params = StdioServerParameters(
        command='python',
        args=['server.py'],
        env={'MCP_DATA_DIR': '/srv/mcp-data'},
    )
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print([t.name for t in tools.tools])
            result = await session.call_tool('read_text', {'rel_path': 'notes.txt'})
            print(result.content)
if __name__ == '__main__':
    asyncio.run(main())

调之前先看 list_tools() 返回的 inputSchema，不要凭猜测拼参数。Server 端也要校验参数，不能假设 Client 一定传对。

3. 权限、审计与提示注入防护

安全不要把希望押在模型「听话」上。MCP 的边界要放在 Server 代码、账号和策略引擎里。

可以用一个简单策略文件：

tools:
  allow:
read_text
get_order_status
  confirm:
send_email
  deny:
write_file
execute_sql
limits:
  max_result_bytes: 65536
  max_calls_per_minute: 60

审计日志至少留这些字段：ts、session_id、client_id、user_id、server、tool、args_hash、decision、error、duration_ms、result_bytes。参数里可能有手机号、地址、token，不要全量落盘；存摘要或按字段脱敏。数据库审计和 MCP 审计要对得上，否则出了越权查询查不到是谁发起。

如果 Server 走 Streamable HTTP，OAuth 2.1 的 token 要绑定资源受众，不要把给 A Server 的 token 拿去调 B Server。stdio 场景没有网络认证，但进程用户权限就是边界：用专用低权限用户跑 Server，数据目录只读挂载。

4. 与 LangChain 和 Agent 框架集成

LangChain 侧用 langchain-mcp-adapters 把 MCP 工具转成 LangChain Tool。LangGraph 的 create_react_agent 可以直接接这些工具。下面用环境变量放远程 Server 地址和 token，避免把密钥写进代码。

import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
client = MultiServerMCPClient({
    'files': {
        'command': 'python',
        'args': ['server.py'],
        'transport': 'stdio',
    },
    'orders': {
        'url': os.environ['MCP_ORDERS_URL'],
        'transport': 'streamable_http',
        'headers': {'Authorization': 'Bearer ' + os.environ['MCP_ORDERS_TOKEN']},
    },
})
tools = await client.get_tools()
agent = create_react_agent(ChatOpenAI(model='gpt-4.1'), tools)
result = await agent.ainvoke({
    'messages': [{'role': 'user', 'content': '查订单 A123 的状态'}]
})
print(result['messages'][-1].content)

这段代码里有两个容易踩的坑。第一，get_tools() 会按 Server 返回的工具描述生成 schema；如果某个 Server 暴露了几十个工具，Agent 的选择空间会变大，误调用概率上升。生产环境按角色拆 Client，客服 Agent 只挂文件 Server 和订单只读 Server。第二，远程 Server 的 HTTP 头不要在客户端硬编码，走密钥管理或短时 token。

Agent 框架不只是 LangChain。AutoGen、CrewAI、OpenAI Agents SDK 等也能接 MCP，接法通常分两类：一类把 MCP 工具注册成框架原生工具；另一类让 Agent 通过 Client 会话直接调用。选型时先看框架是否支持 Streamable HTTP、是否支持工具调用确认、是否能把工具输出标记为不可信。

5. 生产部署与监控

本地能跑不等于生产能用。MCP Server 一旦挂到远程，就等于对外开放了一组可调用函数。部署时把下面清单过一遍：

• 固定 MCP SDK、Server 依赖版本，提交 lock 文件
• 容器用非 root 用户，根文件系统只读，只挂载数据目录
• 远程 Server 走 TLS，Streamable HTTP 放在 API 网关后，启用 OAuth 2.1
• 数据库用只读账号，限制连接数、查询超时、返回行数
• 每个工具设置调用超时、结果大小上限、每用户限流
• 高风险工具加人工确认，拒绝策略写进代码而不是提示词
• 日志脱敏，审计日志单独留存，设置访问权限
• 保留上一个镜像和配置，支持灰度与快速回滚

远程部署还有一个细节：Streamable HTTP 的会话可能跨多个请求。网关要支持 sticky session 或让 Server 无状态化；如果做不到，水平扩容时会频繁初始化。stdio 不需要这些，但每个 Host 实例会拉起自己的 Server 子进程，内存和文件句柄要算进容量。

视频学习可以在 B 站、腾讯视频、优酷、抖音搜「Anthropic MCP」「Model Context Protocol 教程」，优先看官方账号或带官方文档回链的内容。免责声明：本文不附具体视频链接，也不与视频作者有利益关系；视频内容可能滞后于规范，以 MCP 官方规范 为准。

今天能做的一件事

选一个只读数据源，比如项目里的 docs/ 目录，按第 2 节写一个只允许读该目录的 Server。用官方 Inspector 验证工具列表和参数 schema：

npx @modelcontextprotocol/inspector python server.py

先确认越权路径被拒绝，再接进 Agent。别一上来就连生产库。