资讯动态

DeepSeek-Agent-Harness-2026终极指南-第13章第62节-接入MCP万亿生态-自己写一个MCPServer并发布

发布时间:2026/10/5 6:52:59 来源:尧图企业网站定制
DeepSeek Agent Harness 2026终极指南 - 第13章第62节 自己写一个MCP Server并发布第61节能用别人的 MCP Server 了但更见功力的是写一个自己的 Server。这节用官方 Python SDK 写一个数据库查询 MCP Servertools/resources 双原语、FastMCP 装饰器开发、本地调试与发布清单。写完它别人的 AgentClaude、Cursor、我们的 DeepPilot都能用你的工具。本文导航为什么写 Server 是 MCP 生态的关键FastMCP官方 Python SDK 上手tools 与 resources 双原语完整实现db_server.py本地调试连上自己的 Server发布清单小结为什么写 Server 是 MCP 生态的关键MCP 生态的价值是双向的既有人用工具也得有人造工具。用工具第60、61节连接别人的 Server用现成能力。造工具本节写自己的 Server把公司内部能力数据库、API、业务系统暴露给 AI。对公司团队来说写 Server 的价值尤其大内部系统 AI 化把内部数据库、API 包装成 MCP Server让 AI 助手能直接查业务数据。跨工具复用写一个 ServerClaude、Cursor、DeepPilot 都能用不用为每个工具单独适配。能力沉淀把团队的领域知识封装成工具新人也能直接调用。FastMCP官方 Python SDK 上手官方 Python SDK 提供FastMCP用装饰器就能定义工具# 安装官方 SDKpipinstallmcpfrommcp.server.fastmcpimportFastMCP# 创建 Server 实例mcpFastMCP(db-query)mcp.tool()defquery(sql:str)-str:执行 SQL 查询返回结果。# 这里连接真实数据库return查询结果...mcp.tool()装饰器自动注册工具到 Server从函数签名和 docstring 生成 JSON Schema处理 tools/list 和 tools/call 请求这跟第36节我们自己写的tool装饰器几乎一模一样——好东西的思路都是相通的。tools 与 resources 双原语第59节讲过三大原语tools/resources/prompts这里实现其中两个tools可执行操作查数据库、执行 SQLmcp.tool()defquery(sql:str)-str:执行 SQL 查询语句返回结果集。importsqlite3 connsqlite3.connect(mydb.db)cursorconn.cursor()cursor.execute(sql)rowscursor.fetchall()conn.close()returnstr(rows)resources可读取数据读数据库表结构、schemamcp.resource(db://schema)defget_schema()-str:获取数据库表结构只读。return表结构users(id, name, email)...区别tool有副作用能执行写操作需要模型明确调用resource无副作用只读模型可以自由读取作为上下文完整实现db_server.py写一个完整的数据库查询 Server# db_server.py —— 数据库查询 MCP Serverfrom__future__importannotationsimportsqlite3frommcp.server.fastmcpimportFastMCP# 创建 ServermcpFastMCP(db-query)# tools 原语可执行操作 mcp.tool()defquery(sql:str)-str: 执行 SQL 查询语句返回结果集。 参数 - sql: 要执行的 SQL 查询语句只支持 SELECT避免写操作 # 安全检查只允许 SELECTifnotsql.strip().lower().startswith(select):return错误只允许 SELECT 查询禁止写操作connsqlite3.connect(mydb.db)cursorconn.cursor()try:cursor.execute(sql)rowscursor.fetchall()# 格式化为可读文本ifnotrows:return查询结果为空return\n.join(str(row)forrowinrows)exceptExceptionase:returnf查询失败:{e}finally:conn.close()mcp.tool()deflist_tables()-str:列出数据库中的所有表。connsqlite3.connect(mydb.db)cursorconn.cursor()cursor.execute(SELECT name FROM sqlite_master WHERE typetable)tablescursor.fetchall()conn.close()return\n.join(t[0]fortintables)# resources 原语可读取数据 mcp.resource(db://schema/{table})defget_table_schema(table:str)-str:获取指定表的字段结构。connsqlite3.connect(mydb.db)cursorconn.cursor()cursor.execute(fPRAGMA table_info({table}))columnscursor.fetchall()conn.close()return\n.join(f{c[1]}({c[2]})forcincolumns)# 运行 Serverstdio 模式if__name____main__:mcp.run()几个设计要点安全校验query工具只允许 SELECT禁止写操作呼应第10章的安全思想。资源可读get_table_schema用 resource 暴露表结构模型可自由读取。参数化资源db://schema/{table}支持路径参数。stdio 模式mcp.run()默认 stdio 传输跟第60节手写客户端对接。本地调试连上自己的 Server写完 Server用第60节的手写客户端连上它uv run python-c from deep_pilot.mcp_client import MCPClient # 连上自己的 db-query Server client MCPClient([python, db_server.py]) client.initialize() # 发现工具 tools client.list_tools() print(f发现 {len(tools)} 个工具:) for t in tools: print(f - {t[\name\]}: {t[\description\][:30]}) # 调用工具 result client.call_tool(list_tables, {}) print(f数据库表: {result}) client.close() 控制台输出2026-09-14 15:00:01 | INFO | mcp_client | 握手成功Server: db-query 发现 2 个工具: - query: 执行 SQL 查询语句 - list_tables: 列出数据库中的所有表 调用结果: users, orders, products自己的 Server 跑通了——发现 2 个工具调用成功。发布清单写完 Server 要发布让别人能用1. 打包成可安装包pipinstall-e.# 本地安装# 或发布到 PyPIpython-mbuild twine upload dist/*2. 提供配置文件让 MCP 客户端能发现并连接{mcpServers:{db-query:{command:python,args:[db_server.py],env:{DB_PATH:./mydb.db}}}}3. 写 README说明工具、参数、示例4. 发布到 MCP 目录官方 registry让全世界发现小结写 Server 是生态关键既用又造才能把内部能力 AI 化、跨工具复用。FastMCP 上手mcp.tool()装饰器自动注册 生成 Schema跟我们的 tool 思路一致。双原语tools可执行有副作用 resources可读无副作用。安全校验query 工具只允许 SELECT延续第10章的最小权限思想。本地调试用第60节的手写客户端连自己的 Server发现工具、调用成功。发布清单打包、配置文件、README、发布到官方目录。DeepPilot v0.8 MCP 双向打通——既能用别人的 Server也能写自己的 Server。下节预告第13章 MCP 收官了。工具生态接入完毕但 Agent 的性能和成本还有优化空间——同样的功能高手写出来更快更省。下一节进入第14章高级实战性能、成本与硬核优化先做性能优化连接池复用、API 预热、并发调用、首 token 延迟优化附优化前后对比数据。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑