AI 函数调用与 MCP 协议深度解析：从 OpenAI tools 到 Model Context Protocol

1
from openai import OpenAI
2
import json
3

4
client = OpenAI()
5

6
tools = [
7
    {
8
        "type": "function",
9
        "function": {
10
            "name": "get_weather",
11
            "description": "查询指定城市的实时天气",
12
            "parameters": {
13
                "type": "object",
14
                "properties": {
15
                    "city": {"type": "string", "description": "城市名称"},
16
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
17
                },
18
                "required": ["city"]
19
            }
20
        }
21
    }
22
]
23

24
resp = client.chat.completions.create(
25
    model="gpt-4o",
26
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}],
27
    tools=tools,
28
    tool_choice="auto"  # "auto" | "none" | {"type":"function","function":{"name":"get_weather"}}
29
)
30

31
msg = resp.choices[0].message
32
if msg.tool_calls:
33
    for tc in msg.tool_calls:
34
        args = json.loads(tc.function.arguments)
35
        print(f"模型想调用 {tc.function.name}({args})")
36
        # → 模型想调用 get_weather({'city': '北京', 'unit': 'celsius'})

1
import anthropic
2
import json
3

4
client = anthropic.Anthropic()
5

6
tools = [
7
    {
8
        "name": "get_weather",
9
        "description": "查询指定城市的实时天气",
10
        "input_schema": {  # ← 注意字段名不同
11
            "type": "object",
12
            "properties": {
13
                "city": {"type": "string", "description": "城市名称"},
14
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
15
            },
16
            "required": ["city"]
17
        }
18
    }
19
]
20

21
resp = client.messages.create(
22
    model="claude-3-5-sonnet-20241022",
23
    max_tokens=1024,
24
    tools=tools,
25
    messages=[{"role": "user", "content": "北京今天天气怎么样？"}]
26
)
27

28
for block in resp.content:
29
    if block.type == "tool_use":  # ← 块类型不同
30
        print(f"模型想调用 {block.name}({block.input})")
31
        # → 模型想调用 get_weather({'city': '北京', 'unit': 'celsius'})

1
import google.generativeai as genai
2

3
genai.configure(api_key="YOUR_KEY")
4

5
weather_fn = {
6
    "name": "get_weather",
7
    "description": "查询指定城市的实时天气",
8
    "parameters": {
9
        "type": "object",
10
        "properties": {
11
            "city": {"type": "string"},
12
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
13
        },
14
        "required": ["city"]
15
    }
16
}
17

18
model = genai.GenerativeModel(
19
    model_name="gemini-1.5-pro",
20
    tools=[weather_fn]
21
)
22

23
chat = model.start_chat()
24
resp = chat.send_message("北京今天天气怎么样？")
25

26
for part in resp.parts:
27
    if fn := part.function_call:
28
        print(f"模型想调用 {fn.name}({dict(fn.args)})")

1
┌──────────────────────────────────────────────────────────┐
2
│                       MCP 生态                             │
3
├──────────────────────────────────────────────────────────┤
4
│                                                          │
5
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐ │
6
│  │ Cursor   │  │  Claude  │  │ Continue │  │   Zed    │ │
7
│  │  (MCP    │  │ Desktop  │  │  (MCP    │  │  (MCP    │ │
8
│  │ Client)  │  │ (Client) │  │ Client)  │  │ Client)  │ │
9
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘ │
10
│       │             │              │             │       │
11
│       └─────────────┴──────┬───────┴─────────────┘       │
12
│                            │ MCP 协议（JSON-RPC 2.0）      │
13
│                ┌───────────┴───────────┐                  │
14
│                │                       │                  │
15
│          ┌─────▼─────┐         ┌──────▼──────┐            │
16
│          │  Filesystem│         │  GitHub     │            │
17
│          │   Server   │         │   Server    │            │
18
│          └───────────┘         └─────────────┘            │
19
│          ┌───────────┐         ┌─────────────┐            │
20
│          │ Postgres  │         │   Brave     │            │
21
│          │   Server  │         │   Search    │            │
22
│          └───────────┘         └─────────────┘            │
23
└──────────────────────────────────────────────────────────┘

1
"""
2
MCP Server: 代码评审助手
3
功能：
4
- Tool: review_code（评审一段 Python 代码）
5
- Tool: check_security（检测常见安全问题）
6
- Resource: review://rules（评审规则文档）
7
- Prompt: review-pr（生成 PR 评审 prompt 模板）
8
"""
9

10
import asyncio
11
import ast
12
import re
13
from mcp.server import Server
14
from mcp.server.stdio import stdio_server
15
from mcp.types import (
16
    Tool, TextContent, ImageContent, EmbeddedResource,
17
    Prompt, PromptMessage, PromptArgument,
18
    Resource,
19
)
20

21
# ============ 1. 创建 MCP Server ============
22
app = Server("code-reviewer")
23

24

25
# ============ 2. 列出可用工具 ============
26
@app.list_tools()
27
async def list_tools() -> list[Tool]:
28
    return [
29
        Tool(
30
            name="review_code",
31
            description="对一段 Python 代码进行多维度评审，返回结构化问题列表",
32
            inputSchema={
33
                "type": "object",
34
                "properties": {
35
                    "code": {"type": "string", "description": "待评审的 Python 源码"},
36
                    "strictness": {
37
                        "type": "string",
38
                        "enum": ["loose", "normal", "strict"],
39
                        "default": "normal",
40
                        "description": "评审严格度"
41
                    }
42
                },
43
                "required": ["code"]
44
            }
45
        ),
46
        Tool(
47
            name="check_security",
48
            description="扫描代码中的常见安全问题（SQL 注入、硬编码密钥、eval、shell 注入等）",
49
            inputSchema={
50
                "type": "object",
51
                "properties": {
52
                    "code": {"type": "string"}
53
                },
54
                "required": ["code"]
55
            }
56
        ),
57
    ]
58

59

60
# ============ 3. 工具实现 ============
61
SECURITY_RULES = [
62
    (r"\beval\s*\(", "使用了 eval()，存在任意代码执行风险"),
63
    (r"\bexec\s*\(", "使用了 exec()，存在任意代码执行风险"),
64
    (r"subprocess\.(?:call|run|Popen).*shell\s*=\s*True", "subprocess 启用了 shell=True，可能导致命令注入"),
65
    (r"os\.system\s*\(", "os.system() 容易受 shell 注入影响，建议用 subprocess.run 并禁用 shell"),
66
    (r"password\s*=\s*['\"][^'\"]+['\"]", "硬编码了 password 字段"),
67
    (r"api_key\s*=\s*['\"][^'\"]+['\"]", "硬编码了 api_key 字段"),
68
    (r"SELECT.*\+.*FROM", "疑似 SQL 拼接，请使用参数化查询"),
69
    (r"\.format\(.*\).*(?:SELECT|INSERT|DELETE|UPDATE)", "疑似通过 .format() 拼接 SQL"),
70
    (r"pickle\.loads?\s*\(", "pickle 反序列化存在 RCE 风险，请改用 json"),
71
]
72

73
STYLE_RULES = [
74
    (r"except\s*:\s*$", "裸 except 会吞掉所有异常，包括 KeyboardInterrupt"),
75
    (r"except\s+Exception\s*:\s*$", "过于宽泛的异常捕获"),
76
    (r"^\s*print\s*\(", "代码中残留 print 调试语句"),
77
    (r"TODO|FIXME|XXX", "存在未完成的 TODO/FIXME"),
78
    (r"def\s+\w+\([^)]*\)\s*->\s*$", "函数签名末尾缺返回类型注解"),
79
]
80

81

82
def _review_python(code: str, strictness: str) -> dict:
83
    """执行评审逻辑"""
84
    issues = []
85
    severities = {"loose": ["high"], "normal": ["high", "medium"], "strict": ["high", "medium", "low"]}
86

87
    # 静态文本扫描
88
    for pattern, msg in SECURITY_RULES:
89
        if re.search(pattern, code, re.IGNORECASE | re.MULTILINE):
90
            issues.append({
91
                "type": "security",
92
                "severity": "high",
93
                "message": msg,
94
                "line": _find_line(code, pattern)
95
            })
96

97
    for pattern, msg in STYLE_RULES:
98
        if strictness in ("normal", "strict") and re.search(pattern, code, re.MULTILINE):
99
            issues.append({
100
                "type": "style",
101
                "severity": "medium",
102
                "message": msg,
103
                "line": _find_line(code, pattern)
104
            })
105

106
    # AST 分析
107
    try:
108
        tree = ast.parse(code)
109
        issues.extend(_ast_analysis(tree, strictness))
110
    except SyntaxError as e:
111
        issues.append({
112
            "type": "syntax",
113
            "severity": "high",
114
            "message": f"语法错误: {e.msg}",
115
            "line": e.lineno
116
        })
117

118
    # 过滤严重等级
119
    issues = [i for i in issues if i["severity"] in severities.get(strictness, ["high", "medium"])]
120

121
    # 计算分数
122
    high = sum(1 for i in issues if i["severity"] == "high")
123
    medium = sum(1 for i in issues if i["severity"] == "medium")
124
    low = sum(1 for i in issues if i["severity"] == "low")
125
    score = max(0, 100 - high * 20 - medium * 5 - low * 1)
126

127
    return {
128
        "score": score,
129
        "summary": f"发现 {len(issues)} 个问题（高:{high} 中:{medium} 低:{low}）",
130
        "issues": issues,
131
        "lines_of_code": len(code.splitlines()),
132
    }
133

134

135
def _ast_analysis(tree: ast.AST, strictness: str) -> list[dict]:
136
    """AST 层面的深度分析"""
137
    issues = []
138
    for node in ast.walk(tree):
139
        # 函数过长
140
        if isinstance(node, ast.FunctionDef):
141
            length = (node.end_lineno or 0) - node.lineno + 1
142
            if length > 50:
143
                issues.append({
144
                    "type": "complexity",
145
                    "severity": "medium",
146
                    "message": f"函数 `{node.name}` 有 {length} 行，建议拆分",
147
                    "line": node.lineno,
148
                })
149
            # 圈复杂度估算（if/for/while/except 数）
150
            complexity = sum(
151
                1 for n in ast.walk(node)
152
                if isinstance(n, (ast.If, ast.For, ast.While, ast.ExceptHandler, ast.BoolOp))
153
            )
154
            if complexity > 10:
155
                issues.append({
156
                    "type": "complexity",
157
                    "severity": "high",
158
                    "message": f"函数 `{node.name}` 圈复杂度过高（≈{complexity}），建议提取子函数",
159
                    "line": node.lineno,
160
                })
161

162
        # 全局可变状态
163
        if isinstance(node, ast.Global):
164
            issues.append({
165
                "type": "anti-pattern",
166
                "severity": "medium",
167
                "message": "使用了 global 语句，可能破坏封装",
168
                "line": node.lineno,
169
            })
170
    return issues
171

172

173
def _find_line(code: str, pattern: str) -> int:
174
    """找到 pattern 第一次出现的行号"""
175
    m = re.search(pattern, code, re.IGNORECASE | re.MULTILINE)
176
    return m.start() // 80 + 1 if m else 0  # 粗略估算
177

178

179
# ============ 4. 工具分发 ============
180
@app.call_tool()
181
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
182
    if name == "review_code":
183
        result = _review_python(arguments["code"], arguments.get("strictness", "normal"))
184
        return [TextContent(type="text", text=_format_review(result))]
185

186
    elif name == "check_security":
187
        issues = []
188
        for pattern, msg in SECURITY_RULES:
189
            if re.search(pattern, arguments["code"], re.IGNORECASE | re.MULTILINE):
190
                issues.append(f"⚠️ 第 { _find_line(arguments['code'], pattern) } 行: {msg}")
191
        text = "未发现安全问题 ✅" if not issues else "\n".join(issues)
192
        return [TextContent(type="text", text=text)]
193

194
    raise ValueError(f"未知工具: {name}")
195

196

197
def _format_review(result: dict) -> str:
198
    out = [f"📊 代码评分: **{result['score']}/100**", result['summary'], ""]
199
    for issue in result['issues']:
200
        emoji = {"high": "🔴", "medium": "🟡", "low": "🟢"}[issue['severity']]
201
        out.append(f"{emoji} [{issue['type']}] 第 {issue['line']} 行: {issue['message']}")
202
    return "\n".join(out)
203

204

205
# ============ 5. Resources（应用读取的数据）============
206
@app.list_resources()
207
async def list_resources() -> list[Resource]:
208
    return [
209
        Resource(
210
            uri="review://rules",
211
            name="评审规则文档",
212
            description="当前 MCP Server 使用的所有评审规则",
213
            mimeType="text/markdown"
214
        )
215
    ]
216

217

218
@app.read_resource()
219
async def read_resource(uri: str) -> str:
220
    if uri == "review://rules":
221
        rules_md = "# 评审规则\n\n## 安全规则\n"
222
        for pattern, msg in SECURITY_RULES:
223
            rules_md += f"- `{pattern}` → {msg}\n"
224
        rules_md += "\n## 风格规则\n"
225
        for pattern, msg in STYLE_RULES:
226
            rules_md += f"- `{pattern}` → {msg}\n"
227
        return rules_md
228
    raise ValueError(f"未知资源: {uri}")
229

230

231
# ============ 6. Prompts（预定义模板）============
232
@app.list_prompts()
233
async def list_prompts() -> list[Prompt]:
234
    return [
235
        Prompt(
236
            name="review-pr",
237
            description="生成 PR 评审 prompt 模板",
238
            arguments=[
239
                PromptArgument(
240
                    name="diff",
241
                    description="PR 的 diff 内容",
242
                    required=True
243
                )
244
            ]
245
        )
246
    ]
247

248

249
@app.get_prompt()
250
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
251
    if name == "review-pr":
252
        return [
253
            PromptMessage(
254
                role="user",
255
                content=TextContent(
256
                    type="text",
257
                    text=f"请评审以下 PR diff，重点关注正确性、安全性、性能：\n\n```diff\n{arguments['diff']}\n```"
258
                )
259
            )
260
        ]
261
    raise ValueError(f"未知 prompt: {name}")
262

263

264
# ============ 7. 启动 Server ============
265
async def main():
266
    async with stdio_server() as (read_stream, write_stream):
267
        await app.run(
268
            read_stream,
269
            write_stream,
270
            app.create_initialization_options()
271
        )
272

273

274
if __name__ == "__main__":
275
    asyncio.run(main())

1
"""
2
Claude Agent SDK + MCP Server：让 Claude 调用我们的代码评审工具
3
"""
4
import anyio
5
from claude_agent_sdk import query, ClaudeAgentOptions
6

7

8
async def main():
9
    options = ClaudeAgentOptions(
10
        model="claude-sonnet-4-5",
11
        system_prompt="你是一个代码评审专家，使用 review_code 工具评审用户提交的代码。",
12
        mcp_servers={
13
            "code-reviewer": {
14
                "command": "python",
15
                "args": ["-m", "mcp_code_reviewer.server"],
16
            }
17
        },
18
        allowed_tools=["review_code", "check_security"],
19
        max_turns=5,
20
    )
21

22
    code_to_review = """
23
def get_user(user_id):
24
    password = "admin123"
25
    query = "SELECT * FROM users WHERE id = " + user_id
26
    result = eval(query)
27
    return result
28
"""
29

30
    async for message in query(
31
        prompt=f"请评审以下代码：\n```python\n{code_to_review}\n```",
32
        options=options,
33
    ):
34
        if hasattr(message, "content"):
35
            print(message.content)
36

37

38
anyio.run(main)