Function Call、MCP 和 Skill——Python Agent 工具调用的三层阶梯

思考 → 行动 → 观察 → 思考 → 行动 → ... → 结束

from openai import OpenAI
import json

client = OpenAI()

# 工具注册表：工具名 → (执行函数, JSON Schema)
TOOL_MAP = {
    "get_weather": (
        lambda city: json.dumps({"city": city, "temp": 26, "condition": "晴"}),
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "description": "查询指定城市的天气",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {"type": "string", "description": "城市名"}
                    },
                    "required": ["city"]
                }
            }
        }
    )
}

def run_agent(user_query):
    # 收集所有工具的 Schema
    tool_schemas = [schema for _, (_, schema) in TOOL_MAP.items()]
    messages = [{"role": "user", "content": user_query}]

    for _ in range(10):
        resp = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tool_schemas
        )
        msg = resp.choices[0].message

        # 没有 tool_calls → LLM 直接回答了
        if not msg.tool_calls:
            return msg.content

        # 把 LLM 的回复加入历史
        messages.append(msg.model_dump())

        # 逐个执行工具调用
        for tc in msg.tool_calls:
            func_name = tc.function.name
            func_args = json.loads(tc.function.arguments)

            if func_name not in TOOL_MAP:
                result = f"工具 {func_name} 不存在"
            else:
                try:
                    fn, _ = TOOL_MAP[func_name]
                    result = fn(**func_args)
                except Exception as e:
                    result = f"工具执行出错: {e}"

            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": str(result)
            })

    return "超过最大轮次"

import asyncio
import json
from mcp import ClientSession
from mcp.client.stdio import stdio_client
from openai import OpenAI

client = OpenAI()

async def run_mcp_agent(user_query, mcp_server_command):
    # 启动 MCP Server（本地命令行方式）
    async with stdio_client(mcp_server_command) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 自动发现 Server 提供的所有工具
            mcp_tools = await session.list_tools()
            print(f"发现 {len(mcp_tools.tools)} 个工具: {[t.name for t in mcp_tools.tools]}")

            # 把 MCP 工具的 Schema 转成 OpenAI 格式
            openai_tools = []
            for t in mcp_tools.tools:
                openai_tools.append({
                    "type": "function",
                    "function": {
                        "name": t.name,
                        "description": t.description or "",
                        "parameters": t.inputSchema
                    }
                })

            # Agent 循环（和 Function Call 的循环逻辑一样）
            messages = [{"role": "user", "content": user_query}]

            for _ in range(10):
                resp = client.chat.completions.create(
                    model="gpt-4o",
                    messages=messages,
                    tools=openai_tools
                )
                msg = resp.choices[0].message

                if not msg.tool_calls:
                    return msg.content

                messages.append(msg.model_dump())

                for tc in msg.tool_calls:
                    func_name = tc.function.name
                    func_args = json.loads(tc.function.arguments)

                    # 通过 MCP Client 执行工具，而不是自己写函数
                    result = await session.call_tool(func_name, func_args)

                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": str(result.content)
                    })

            return "超过最大轮次"

mcp_server_command = ["python", "weather-server.py"]

class Skill:
    def __init__(self, name, description, tools, system_prompt, rules=None):
        self.name = name
        self.description = description
        self.tools = tools            # 这个 Skill 要用到的工具列表
        self.system_prompt = system_prompt  # 注入到 LLM 上下文的提示
        self.rules = rules or []      # 执行规则/约束

code_review_skill = Skill(
    name="code_review",
    description="对代码做审查，检查语法、风格、安全问题",
    tools=["read_file", "run_lint", "run_test", "grep_search"],
    system_prompt="""你是一个代码审查专家。当用户让你 review 代码时：

1. 先用 read_file 读取文件内容
2. 用 run_lint 检查语法和风格问题
3. 用 grep_search 查找常见安全模式（eval、exec、硬编码密钥等）
4. 如果有测试文件，用 run_test 跑一下测试

审查时要关注：安全性、可读性、性能、错误处理。""",
    rules=[
        "先读文件再下结论",
        "不要修改代码，只给出建议",
        "如果 lint 报错超过 10 个，先让用户处理 lint 问题"
    ]
)

def load_skill_for_task(task, skills):
    # 简单匹配：根据任务描述选择 Skill
    for skill in skills:
        if skill.name in task:
            return skill
    return None

def run_agent_with_skill(user_query, skill):
    messages = [
        {"role": "system", "content": skill.system_prompt},
        {"role": "user", "content": user_query}
    ]

    # 只注册这个 Skill 需要的工具，不是全部
    active_tools = [t for t in ALL_TOOL_SCHEMAS if t["function"]["name"] in skill.tools]

    # 后续的 Agent 循环和之前一样...

Agent 循环
┌─────────────────────────────┐
│  Skill（任务编排层）          │  ← 注入 Prompt + 选择工具子集
│  ┌───────────────────────┐  │
│  │  MCP（工具连接层）      │  │  ← 标准化协议发现/执行工具
│  │  ┌─────────────────┐  │  │
│  │  │ Function Call    │  │  │  ← 单次工具调用的原子操作
│  │  └─────────────────┘  │  │
│  └───────────────────────┘  │
└─────────────────────────────┘

# 3 个工具以内，手写最直接
TOOL_MAP = {
    "get_weather": (get_weather, get_weather_schema),
    "get_time": (get_time, get_time_schema),
}

# 3 个 MCP Server，Agent 代码只负责循环
servers = [
    ["npx", "github-mcp-server"],
    ["npx", "slack-mcp-server"],
    ["npx", "postgres-mcp-server"],
]

class MyAgent:
    def __init__(self):
        self.skills = {
            "code_review": CodeReviewSkill(),
            "github_manage": GitHubSkill(use_mcp=True),
            "doc_writer": DocWriterSkill(),
        }

    def run(self, user_input):
        skill = self.match_skill(user_input)
        # skill.tools 里面可能同时有手写的 Function Call 工具
        # 和通过 MCP Server 连接的远程工具
        return self.agent_loop(user_input, tools=skill.tools,
                               system_prompt=skill.system_prompt)

import requests
from openai import OpenAI
import json

GITHUB_TOKEN = "ghp_xxx"
REPO = "user/repo"
HEADERS = {"Authorization": f"token {GITHUB_TOKEN}"}

def list_issues():
    r = requests.get(
        f"https://api.github.com/repos/{REPO}/issues?state=open",
        headers=HEADERS
    )
    return json.dumps([{"number": i["number"], "title": i["title"]}
                       for i in r.json()[:10]])

def create_issue(title, body=""):
    r = requests.post(
        f"https://api.github.com/repos/{REPO}/issues",
        headers=HEADERS,
        json={"title": title, "body": body}
    )
    data = r.json()
    return json.dumps({"number": data.get("number"), "url": data.get("html_url")})

def close_issue(number):
    r = requests.patch(
        f"https://api.github.com/repos/{REPO}/issues/{number}",
        headers=HEADERS,
        json={"state": "closed"}
    )
    return json.dumps({"number": number, "closed": r.status_code == 200})

# 手写三份 Schema
def make_tool_schema(name, desc, props, required):
    return {"type": "function", "function": {
        "name": name, "description": desc,
        "parameters": {"type": "object", "properties": props,
                       "required": required}
    }}

TOOL_MAP = {
    "list_issues":  (list_issues,  make_tool_schema(
        "list_issues", "列出仓库 open 状态的 Issues", {}, [])),
    "create_issue": (create_issue, make_tool_schema(
        "create_issue", "创建一个新 Issue",
        {"title": {"type": "string", "description": "标题"},
         "body":  {"type": "string", "description": "内容"}},
        ["title"])),
    "close_issue":  (close_issue,  make_tool_schema(
        "close_issue", "关闭指定 Issue",
        {"number": {"type": "integer", "description": "Issue 编号"}},
        ["number"])),
}

import asyncio
import json
from mcp import ClientSession
from mcp.client.stdio import stdio_client
from openai import OpenAI

client = OpenAI()

async def run_github_agent(user_query):
    # 启动 GitHub MCP Server（社区现成的）
    server_cmd = {
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"],
        "env": {"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxx"}
    }

    async with stdio_client(server_cmd) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # 自动发现所有 GitHub 工具
            tools = await session.list_tools()
            print(f"拿到 {len(tools.tools)} 个 GitHub 工具")

            # 转成 OpenAI 格式
            openai_tools = [{"type": "function", "function": {
                "name": t.name, "description": t.description or "",
                "parameters": t.inputSchema
            }} for t in tools.tools]

            messages = [{"role": "user", "content": user_query}]

            for _ in range(10):
                resp = client.chat.completions.create(
                    model="gpt-4o", messages=messages, tools=openai_tools)
                msg = resp.choices[0].message
                if not msg.tool_calls:
                    return msg.content
                messages.append(msg.model_dump())
                for tc in msg.tool_calls:
                    args = json.loads(tc.function.arguments)
                    result = await session.call_tool(tc.function.name, args)
                    messages.append({"role": "tool", "tool_call_id": tc.id,
                                     "content": str(result.content)})
            return "max turns"

github_issues_skill = Skill(
    name="github_issues",
    description="管理 GitHub Issues",
    tools=["github_list_issues", "github_create_issue", "github_close_issue",
           "github_add_comment"],
    system_prompt="""你是 GitHub Issues 管理助手。当用户提到 Issue 相关操作时：
1. 创建 Issue 前确认标题不为空
2. 关闭 Issue 前先列出当前 open 的 issues，让用户确认编号
3. 操作完成后总结做了什么""",
    rules=[
        "不要关闭用户没提到的 Issue",
        "创建 Issue 时 body 默认为空字符串",
        "如果用户说'我有哪些 issue'，默认列出 open 状态的"
    ]
)

async def run_skill_agent(user_query):
    skill = github_issues_skill

    # 连接 MCP Server，但只注册 Skill 里声明的工具
    async with stdio_client(mcp_server_cmd) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            all_tools = await session.list_tools()

            # 只取 Skill 需要的工具
            filtered = [t for t in all_tools.tools if t.name in skill.tools]

            messages = [
                {"role": "system", "content": skill.system_prompt},
                {"role": "user", "content": user_query}
            ]
            # ... 后续循环相同