本文较长,建议点赞收藏,以免遗失。文末还有福利。关注我,实力宠粉。
一、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_servercd mcp_serveruv venv..venv\Scripts\activate
- 跨平台依赖:
# 核心依赖安装uv add mcp[cli] httpx psycopg2
- Node.js环境(v18.20.8必备):
在 nodejs.org/zh-cn/downl…,选择 v22.17.0(LTS) 版本进行下载并安装:
# 验证安装node --versionnpm --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),))
- 访问控制:
# 生产环境限制访问IPmcp = 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 SimpleConnectionPoolpool = SimpleConnectionPool(5, 20, **DB_CONFIG)# 2. 异步处理@mcp.resource("db://async")async def async_data(): async with pool.getconn() as conn: # 异步查询
结语:MCP的核心价值
通过Resource/Prompt/Tool三层抽象,MCP实现了:
- 安全隔离:工具操作通过沙箱环境执行标准化接口:统一AI模型与外部系统交互规范灵活扩展:支持数据库/API/文件系统等多类连接开发效率:Inspector工具实现可视化调试本文相关MCP实战粉丝自行领取:MCP相关实战代码
如果本次分享对你有所帮助,记得告诉身边有需要的朋友,"我们正在经历的不仅是技术迭代,而是认知革命。当人类智慧与机器智能形成共生关系,文明的火种将在新的维度延续。"在这场波澜壮阔的文明跃迁中,主动拥抱AI时代,就是掌握打开新纪元之门的密钥,让每个人都能在智能化的星辰大海中,找到属于自己的航向。