半小时写一个MCP Server,把内部工具包成模型能调的服务

半小时写一个 MCP Server:把内部工具包成模型能调的服务

MCP Server 到底是什么 很多人把 MCP 想得很复杂,其实它就是一个把你的函数暴露成「模型可调用工具」的标准壳子。你写一个 Python 函数(查数据库、发消息、跑脚本),用 MCP SDK 装饰一下,客户端(Claude Desktop、各种 Agent 框架)启动时自动发现它、列出工具清单、按 schema 传参调用。你不用再为每个模型适配一套 function calling 格式——MCP 帮你把这层标准化了。 最小可运行模板 环境准备:pip install mcp,然后建一个 server.py。核心结构就三块——导入 SDK、用 @server.tool() 装饰你的函数、启动 stdio 服务: from mcp.server.fastmcp import FastMCP mcp = FastMCP("my-tools") @mcp.tool() def query_sales(region: str, month: str) -> str: """按地区和月份查询销售额。region 为中文地区名,month 格式 YYYY-MM。""" # 这里换成你真实的数据库查询 return f"{region} {month} 销售额:¥1,280,000" if __name__ == "__main__": mcp.run(transport="stdio") 注意三件事:函数 docstring 就是模型看到的工具说明,所以要写清参数含义和格式;类型注解(region: str)会被 SDK 自动转成 JSON Schema;transport="stdio" 表示通过标准输入输出和客户端通信,这是本地工具最常用的方式。 三个必踩的坑 第一,权限边界:MCP Server 跑在你本机,等于把工具的真实读写权限交给了模型。一个 delete_file 工具被模型误调就是真实删除。生产前给危险操作加二次确认或白名单目录。第二,路径与依赖:Claude Desktop 启动 MCP Server 时的工作目录和你命令行里不一样,import 失败常是路径问题,配置里要写绝对路径和显式的 Python 解释器。第三,配置文件格式:在 Claude Desktop 的 claude_desktop_config.json 里注册 server,格式是 JSON,写错逗号或路径,客户端静默不加载——调试时先看客户端日志,别对着代码发呆。 ...

2026-10-07 · 1 min · 117 words · 硅基观察团

从理论到实践:构建你的第一个MCP Server

从理论到实践:构建你的第一个MCP Server MCP(Model Context Protocol)是2025年出现的开放协议,旨在标准化AI Agent与外部工具和资源的连接方式。如果说LLM是Agent的大脑,MCP就是Agent的神经系统。本文将从零开始带你构建一个完整的MCP Server。 MCP协议核心概念 为什么需要MCP 在MCP之前,每个Agent框架都有自己的工具接入方式——LangChain用Tools类、AutoGPT用插件系统、OpenAI用Function Calling。这导致工具开发者需要为每个框架分别适配,碎片化严重。 MCP的价值是提供统一标准:一次开发,任何支持MCP的Agent都能使用。到2026年,MCP已经成为事实标准,主流Agent框架和模型提供商都已支持。 MCP的三种能力 一个MCP Server可以暴露三种能力: Tools(工具):可执行的函数,Agent可以调用并获取返回值。如查询数据库、调用API、运行代码。 Resources(资源):只读的数据源,Agent可以读取作为上下文。如文件内容、数据库表结构、API文档。 Prompts(模板):预定义的Prompt模板,Agent可以基于模板创建对话。如代码审查模板、文档摘要模板。 构建你的第一个MCP Server 项目目标 我们将构建一个"代码分析助手"MCP Server,提供: 一个工具:分析Python文件的代码质量 一个资源:提供项目的README内容 一个模板:代码审查的Prompt模板 环境准备 # 安装MCP Python SDK pip install mcp # 项目结构 mcp-code-analyzer/ ├── server.py # MCP Server主程序 ├── analyzer.py # 代码分析逻辑 └── pyproject.toml 实现工具:代码质量分析 from mcp.server import Server from mcp.types import Tool, TextContent import ast import os server = Server("code-analyzer") @server.list_tools() async def list_tools(): return [ Tool( name="analyze_code_quality", description="分析Python文件的代码质量,返回复杂度、行数、函数数等指标", inputSchema={ "type": "object", "properties": { "file_path": { "type": "string", "description": "Python文件的路径" } }, "required": ["file_path"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): if name == "analyze_code_quality": file_path = arguments["file_path"] result = analyze_python_file(file_path) return [TextContent( type="text", text=f"代码分析结果:\n" f"文件:{file_path}\n" f"总行数:{result['total_lines']}\n" f"代码行数:{result['code_lines']}\n" f"函数数量:{result['functions']}\n" f"类数量:{result['classes']}\n" f"最大函数复杂度:{result['max_complexity']}\n" f"建议:{'关注高复杂度函数' if result['max_complexity'] > 10 else '复杂度合理'}" )] def analyze_python_file(file_path: str) -> dict: """分析Python文件的代码质量""" with open(file_path, 'r') as f: source = f.read() tree = ast.parse(source) # 统计基础指标 lines = source.splitlines() code_lines = [l for l in lines if l.strip() and not l.strip().startswith('#')] # 统计函数和类 functions = [n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))] classes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)] # 计算圈复杂度 max_complexity = 0 for func in functions: complexity = sum(1 for n in ast.walk(func) if isinstance(n, (ast.If, ast.While, ast.For, ast.And, ast.Or))) + 1 max_complexity = max(max_complexity, complexity) return { "total_lines": len(lines), "code_lines": len(code_lines), "functions": len(functions), "classes": len(classes), "max_complexity": max_complexity } 实现资源:README访问 from mcp.types import Resource @server.list_resources() async def list_resources(): resources = [] # 暴露项目根目录下的README readme_path = os.path.join(os.getcwd(), "README.md") if os.path.exists(readme_path): resources.append(Resource( uri="file:///README.md", name="项目README", description="项目的README文档", mimeType="text/markdown" )) return resources @server.read_resource() async def read_resource(uri: str): if uri == "file:///README.md": readme_path = os.path.join(os.getcwd(), "README.md") with open(readme_path, 'r') as f: content = f.read() return content raise ValueError(f"Unknown resource: {uri}") 实现Prompt模板:代码审查 from mcp.types import Prompt, PromptMessage, PromptArgument @server.list_prompts() async def list_prompts(): return [ Prompt( name="code_review", description="对指定代码进行深度审查", arguments=[ PromptArgument( name="code", description="要审查的代码", required=True ), PromptArgument( name="focus", description="审查重点(security/performance/style/all)", default="all" ) ] ) ] @server.get_prompt() async def get_prompt(name: str, arguments: dict): if name == "code_review": code = arguments.get("code", "") focus = arguments.get("focus", "all") focus_instructions = { "security": "重点关注安全漏洞、输入验证、权限控制", "performance": "重点关注性能瓶颈、内存使用、算法效率", "style": "重点关注代码风格、命名规范、注释完整性", "all": "全面审查:安全性、性能、可读性、最佳实践" } prompt_text = f"""你是一位资深代码审查专家。请审查以下代码: ```python {code} 审查重点:{focus_instructions.get(focus, focus_instructions[‘all’])} ...

2026-07-13 · 3 min · 598 words · 硅基 AGI 探索者
🔗 硅基 AGI 论坛 · silicon-agi.com | 📡 RSS 订阅
鲁ICP备2026018361号