1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

{
  "name": "get_weather",
  "description": "获取指定城市的天气信息",
  "parameters": {
    "type": "object",
    "properties": {
      "city": {
        "type": "string",
        "description": "城市名称"
      },
      "unit": {
        "type": "string",
        "enum": ["celsius", "fahrenheit"],
        "description": "温度单位"
      }
    },
    "required": ["city"]
  }
}

1

{"tool_choice": {"type": "function", "function": {"name": "get_weather"}}}

1

{"tool_choice": "none"}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

{
  "name": "search_products",
  "description": "在产品数据库中搜索商品",
  "parameters": {
    "type": "object",
    "properties": {
      "query": {
        "type": "string",
        "description": "搜索关键词"
      },
      "category": {
        "type": ["string", "null"],
        "enum": ["electronics", "clothing", "books", null],
        "description": "产品类别"
      },
      "price_max": {
        "type": ["number", "null"],
        "description": "最高价格限制"
      },
      "sort_by": {
        "type": ["string", "null"],
        "enum": ["price", "rating", "newest", null],
        "description": "排序方式；为空时由服务端使用默认排序"
      }
    },
    "required": ["query", "category", "price_max", "sort_by"],
    "additionalProperties": false
  }
}

1

[用途]。当[触发条件]时调用。[返回内容概述]。

1
2

❌ 避免：uuid、256px_image_url、mime_type
✓ 推荐：name、image_url、file_type

1
2
3
4
5
6
7
8
9

{
  "name": "create_ticket",
  "input_schema": {...},
  "input_examples": [
    {"title": "紧急：登录页面 500 错误", "priority": "critical", "labels": ["bug"]},
    {"title": "添加暗色模式支持", "labels": ["feature-request"]},
    {"title": "更新文档"}
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

def validate_params(tool_name, params, schema):
    # 检查必需参数
    for required in schema.get("required", []):
        if required not in params:
            raise ValueError(f"缺少必需参数: {required}")

    # 检查参数类型
    for key, value in params.items():
        expected_type = schema["properties"][key]["type"]
        if not isinstance(value, TYPE_MAP[expected_type]):
            raise TypeError(f"参数 {key} 类型错误")

    # 检查枚举值
    for key, value in params.items():
        if "enum" in schema["properties"][key]:
            if value not in schema["properties"][key]["enum"]:
                raise ValueError(f"参数 {key} 值无效")

1

result = execute_tool(tool_name, params)

1
2
3
4

async def execute_tools(tool_calls):
    tasks = [execute_tool_async(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks)
    return results

1
2
3
4
5
6
7

try:
    result = await asyncio.wait_for(
        execute_tool(tool_name, params),
        timeout=30.0
    )
except asyncio.TimeoutError:
    result = {"error": "执行超时"}

1
2
3
4
5
6
7

{
  "status": "success",
  "data": {
    "temperature": 25,
    "condition": "晴"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

{
  "query": "北京天气",
  "result": {
    "city": "北京",
    "temperature": 25,
    "condition": "晴",
    "humidity": 45
  },
  "data_time": "2026-03-01 10:00"
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

{
  "role": "assistant",
  "content": null,
  "tool_calls": [{
    "id": "call_abc123",
    "type": "function",
    "function": {
      "name": "get_weather",
      "arguments": "{\"city\": \"北京\"}"
    }
  }]
},
{
  "role": "tool",
  "tool_call_id": "call_abc123",
  "content": "{\"temperature\": 25, \"condition\": \"晴\"}"
}

1
2
3
4
5
6
7
8
9

{
  "tools": [
    {"type": "tool_search_tool", "name": "tool_search"},
    {
      "name": "github.createPullRequest",
      "defer_loading": true
    }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# 模型只生成声明式计划；服务端负责鉴权、执行、脱敏和审计
workflow = {
    "operation": "expense_budget_check",
    "team": "engineering",
    "period": "Q3",
    "output": "over_budget_summary",
}

result = await workflow_runner.run(
    workflow,
    caller=user_identity,
    policy_checks=["team_scope", "read_only", "pii_redaction"],
    network_scope="internal_tools_only",
)
return result.redacted_summary

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

async def run_analysis_agent(user_query, context):
    messages = context + [{"role": "user", "content": user_query}]

    # 1. 规划阶段
    plan = await llm.plan(messages)
    # output: 1. read_csv, 2. analyze_trend, 3. plot_chart

    results = {}

    for step in plan.steps:
        # 2. 执行阶段
        tool_name = step.tool
        tool_args = step.args

        try:
            # 3. 结果处理
            # 仅截取前5行数据作为上下文，避免 Token 爆炸
            raw_result = await tools.execute(tool_name, tool_args)
            processed_result = truncate_dataframe(raw_result)

            # 注入当前步骤结果到上下文
            messages.append({
                "role": "tool",
                "tool_call_id": step.tool_call_id,
                "content": processed_result
            })

        except Exception as e:
            # 错误恢复机制
            messages.append({"role": "system", "content": f"工具执行失败: {str(e)}，请尝试其他方法"})
            continue

    # 4. 生成最终回复
    final_response = await llm.chat(messages)
    return final_response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "name": "read_file",
  "description": "读取指定路径的文件内容",
  "inputSchema": {
    "type": "object",
    "properties": {
      "path": {"type": "string"}
    },
    "required": ["path"]
  }
}

1
2
3
4
5

{
  "uri": "file://config.json",
  "name": "配置文件",
  "mimeType": "application/json"
}

1
2
3
4
5
6
7

{
  "name": "code_review",
  "description": "代码审查提示模板",
  "arguments": [
    {"name": "code", "required": true}
  ]
}

1
2
3
4
5
6
7
8

{
  "taskId": "task-123",
  "status": "working",
  "createdAt": "2026-05-20T10:30:00Z",
  "lastUpdatedAt": "2026-05-20T10:40:00Z",
  "ttl": 60000,
  "pollInterval": 5000
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

from mcp.server import MCPServer

server = MCPServer("my-service")

@server.tool()
async def my_tool(param: str) -> str:
    """工具描述"""
    result = process(param)
    return result

@server.resource("config://main")
async def get_config() -> str:
    """获取配置"""
    return load_config()

if __name__ == "__main__":
    server.run()

1

用户查询 → 检索系统 → 向量库 → 上下文

1
2
3
4
5
6

用户查询 → 查询规划 → MCP 服务发现 → 多个Resources
                        ├─ 文件系统 (file://)
                        ├─ 数据库 (db://)
                        ├─ API (http://)
                        └─ 实时数据源 (stream://)
                        → 动态上下文组装

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# 示意性伪代码：不同 MCP 客户端的具体方法名会不同
class MCPContextAggregator:
    """使用 MCP 聚合来自多个源的上下文"""

    def __init__(self, mcp_servers: dict):
        """
        mcp_servers: {
            'filesystem': MCPClient(...),
            'database': MCPClient(...),
            'api': MCPClient(...)
        }
        """
        self.servers = mcp_servers

    async def gather_context(self, query: str, context_type: str) -> str:
        """
        根据查询类型从不同的MCP服务获取上下文
        """
        context_parts = []

        # 优先级1：从数据库获取结构化数据
        if 'database' in self.servers:
            db_context = await self.servers['database'].read_resource(
                f"db://query/{context_type}"
            )
            if db_context:
                context_parts.append(f"【结构化数据】\n{db_context}")

        # 优先级2：从文件系统获取文档
        if 'filesystem' in self.servers:
            file_context = await self.servers['filesystem'].read_resource(
                f"file://docs/{context_type}/"
            )
            if file_context:
                context_parts.append(f"【文档内容】\n{file_context}")

        # 优先级3：从API获取实时数据
        if 'api' in self.servers:
            api_context = await self.servers['api'].read_resource(
                f"http://api/context?type={context_type}"
            )
            if api_context:
                context_parts.append(f"【实时数据】\n{api_context}")

        return "\n---\n".join(context_parts)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

# 示意性伪代码：用于说明如何把 MCP Tools 作为上下文处理算子
class MCPContextManipulator:
    """使用MCP工具在推理过程中变换上下文"""

    def __init__(self, mcp_tools: dict):
        """
        mcp_tools: {
            'compress': Tool,
            'filter': Tool,
            'enrich': Tool,
            'translate': Tool
        }
        """
        self.tools = mcp_tools

    async def compress_context(self, context: str, ratio: float = 0.5) -> str:
        """使用MCP工具压缩上下文"""
        result = await self.tools['compress'].execute(
            context=context,
            compression_ratio=ratio
        )
        return result['compressed']

    async def filter_context(self, context: str, keywords: list) -> str:
        """过滤上下文，仅保留相关部分"""
        result = await self.tools['filter'].execute(
            context=context,
            keep_keywords=keywords
        )
        return result['filtered']

    async def enrich_context(self, context: str, knowledge_bases: list) -> str:
        """使用知识库丰富上下文"""
        result = await self.tools['enrich'].execute(
            context=context,
            sources=knowledge_bases
        )
        return result['enriched']

    async def translate_context(self, context: str, target_language: str) -> str:
        """将上下文翻译为目标语言"""
        result = await self.tools['translate'].execute(
            text=context,
            target_lang=target_language
        )
        return result['translated']

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

# 示意性伪代码：模板获取与渲染接口因客户端而异
class MCPPromptManager:
    """使用MCP Prompts标准化上下文呈现"""

    def __init__(self, mcp_prompt_server):
        """
        mcp_prompt_server: 提供预定义提示词模板的MCP服务器
        """
        self.prompt_server = mcp_prompt_server

    async def build_rag_prompt(self,
                              query: str,
                              retrieved_docs: list,
                              system_context: str = None) -> list:
        """
        使用标准化的RAG提示词模板构建提示

        MCP服务器可提供以下模板：
        - rag/basic: 基础RAG模板
        - rag/cite: 带引文的RAG
        - rag/reasoning: 带推理步骤的RAG
        """

        # 获取结构化 prompt 结果
        prompt_result = await self.prompt_server.get_prompt(
            name="rag/reasoning",
            arguments={
                "system_context": system_context or "你是一个有帮助的AI助手",
                "documents": "\n".join([f"【{doc['id']}】{doc['content']}"
                                       for doc in retrieved_docs]),
                "query": query,
            }
        )

        return prompt_result.messages

    async def build_agent_prompt(self,
                                task: str,
                                tools: list,
                                state: dict) -> list:
        """
        为智能体构建标准化的提示词，包含：
        - 任务描述
        - 可用工具列表
        - 当前状态
        - 推理指导
        """

        prompt_result = await self.prompt_server.get_prompt(
            name="agent/planning",
            arguments={
                "task": task,
                "tools": "\n".join([f"- {tool['name']}: {tool['description']}"
                                    for tool in tools]),
                "current_state": str(state),
                "reasoning_guide": "分步骤思考，先规划再执行",
            }
        )

        return prompt_result.messages

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157

from mcp.server import MCPServer
import hashlib
import os
import re
import httpx
from typing import Optional

# 初始化 MCP Server
mcp = MCPServer("github-server")
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]

SENSITIVE_PATTERNS = [
    re.compile(r"(?i)(api[_-]?key|token|secret|password)\s*[:=]\s*[^\s]+"),
    re.compile(r"ghp_[A-Za-z0-9_]{20,}"),
]

def redact_sensitive(text: str) -> str:
    """只用于确认预览；真实提交内容仍按原文发送。"""
    redacted = text
    for pattern in SENSITIVE_PATTERNS:
        redacted = pattern.sub("<REDACTED>", redacted)
    return redacted

# ============ Tools：可执行的操作 ============

@mcp.tool()
async def search_repos(query: str, language: str = None) -> str:
    """搜索 GitHub 仓库

    Args:
        query: 搜索关键词
        language: 可选，编程语言过滤
    """
    search_query = f"{query} language:{language}" if language else query
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.github.com/search/repositories",
            params={"q": search_query, "per_page": 5},
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}
        )
        repos = resp.json().get("items", [])
        return "\n".join([
            f"- {r['full_name']} ⭐{r['stargazers_count']}: {r['description']}"
            for r in repos
        ])

@mcp.tool()
async def create_issue(
    repo: str,
    title: str,
    body: str,
    dry_run: bool = True,
    confirmed_operation_id: Optional[str] = None,
) -> str:
    """在指定仓库创建 Issue

    Args:
        repo: 仓库全名，如 owner/repo
        title: Issue 标题
        body: Issue 内容
        dry_run: 默认只返回待确认预览，不执行写操作
        confirmed_operation_id: 人工确认后的操作编号
    """
    allowed_repos = set(os.environ.get("GITHUB_ALLOWED_REPOS", "").split(","))
    if repo not in allowed_repos:
        return "创建失败：仓库不在 allowlist 中"
    if len(body) > 4000:
        return "创建失败：Issue 内容过长，请先人工复核"

    operation_id = hashlib.sha256(f"{repo}\0{title}\0{body}".encode()).hexdigest()[:12]
    if dry_run or confirmed_operation_id != operation_id:
        body_preview = redact_sensitive(body)[:500]
        return (
            "待确认的 Issue 创建请求：\n"
            f"- repo: {repo}\n"
            f"- title: {title}\n"
            f"- body_preview: {body_preview}\n"
            f"- operation_id: {operation_id}\n"
            "确认后再以 dry_run=False 且 confirmed_operation_id 等于该编号重试。"
        )

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"https://api.github.com/repos/{repo}/issues",
            json={"title": title, "body": body},
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}
        )
        if resp.status_code == 201:
            return f"Issue 创建成功: {resp.json()['html_url']}"
        return f"创建失败：GitHub API 返回状态码 {resp.status_code}"

# ============ Resources：可读取的数据 ============

@mcp.resource("github://repos/{owner}/{repo}/readme")
async def get_readme(owner: str, repo: str) -> str:
    """获取仓库的 README 内容"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"https://api.github.com/repos/{owner}/{repo}/readme",
            headers={
                "Authorization": f"Bearer {GITHUB_TOKEN}",
                "Accept": "application/vnd.github.raw"
            }
        )
        return resp.text if resp.status_code == 200 else "README 未找到"

@mcp.resource("github://user/starred")
async def get_starred() -> str:
    """获取用户 Star 的仓库列表"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.github.com/user/starred",
            headers={"Authorization": f"Bearer {GITHUB_TOKEN}"}
        )
        repos = resp.json()[:10]
        return "\n".join([f"- {r['full_name']}" for r in repos])

# ============ Prompts：预定义模板 ============

@mcp.prompt()
def code_review_prompt(repo: str, pr_number: int) -> str:
    """生成代码审查的提示词模板"""
    return f"""请审查 {repo} 仓库的 PR #{pr_number}。

审查要点：
1. 代码逻辑是否正确
2. 是否有潜在的性能问题
3. 代码风格是否符合规范
4. 是否有足够的测试覆盖

请先获取 PR 详情，然后逐一分析。"""

@mcp.prompt()
def issue_template(title: str) -> str:
    """生成 Issue 创建模板"""
    return f"""请创建一个规范的 GitHub Issue。

标题: {title}

请按以下格式填写内容：
## 问题描述
[简述问题]

## 复现步骤
1.
2.

## 期望行为
[描述期望的正确行为]

## 环境信息
- 操作系统:
- 版本号:"""

# 运行服务
if __name__ == "__main__":
    mcp.run()

1

pip install mcp httpx

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

{
  "mcpServers": {
    "github": {
      "command": "python",
      "args": ["/path/to/github_server.py"],
      "env": {
        "GITHUB_TOKEN": "${GITHUB_TOKEN}"
      }
    }
  }
}

1
2
3
4
5
6

用户: 帮我搜索 Python 的 RAG 相关项目
助手: [调用 search_repos(query="RAG", language="python")]
       找到以下项目：
       - langchain-ai/langchain ⭐75000: LLM 应用开发框架
       - run-llama/llama_index ⭐30000: 数据框架连接 LLM
       ...