StudyAgent

📚 StudyHelper — 一个 Agent 框架教学项目

> 如果你正在学习 Claude Code 的 harness 框架源码，或者想理解 Function Call + Skills + MCP 三者如何共同构成一个完整 Agent —— 这个项目就是为你准备的。

---

目录

• [1. 这个项目是什么](#1-这个项目是什么)
• [2. Agent 框架全景：Harness 架构深度解析](#2-agent-框架全景harness-架构深度解析)
• [2.1 整体架构图](#21-整体架构图)
• [2.2 主循环（ReAct Loop）详解](#22-主循环react-loop详解)
• [2.3 消息流转全链路](#23-消息流转全链路)
• [2.4 Harness 核心：Tool Registry 模式](#24-harness-核心tool-registry-模式)
• [3. 三层能力体系的定位与关系](#3-三层能力体系的定位与关系)
• [4. 教学：如何添加自己的 Tool](#4-教学如何添加自己的-tool)
• [5. 教学：如何添加自己的 Skill](#5-教学如何添加自己的-skill)
• [6. 教学：如何接入 MCP Server](#6-教学如何接入-mcp-server)
• [6.1 MCP 协议快速理解](#61-mcp-协议快速理解)
• [6.2 配置一个新的 MCP Server](#62-配置一个新的-mcp-server)
• [6.3 MCP 客户端的核心实现](#63-mcp-客户端的核心实现)
• [6.4 排错指南：MCP 连不上的常见原因](#64-排错指南mcp-连不上的常见原因)
• [7. 快速开始](#7-快速开始)
• [8. 项目结构](#8-项目结构)
• [9. 进阶：如何改造为通用的 Harness 框架](#9-进阶如何改造为通用的-harness-框架)

---

1. 这个项目是什么

StudyHelper 是一个知识学习智能体，用户输入想学的主题，Agent 自动完成：

用户: "我想学微积分"
  → 🔍 搜索网页资料
  → 📥 抓取文章正文
  → 📝 编撰分章节教学教材
  → ✍️ 为每章生成练习题（选择/填空/简答/实践）
  → 💾 保存为 Markdown 文件

但它更重要的角色是作为 Agent 框架的教学蓝图。 这个项目刻意把 Function Call、Skills、MCP 三个概念拆解为独立层，让你能看清楚每一层是怎么"插"进 Agent 的，以及它们之间的数据如何流动。

如果你在学 Claude Code 的源码，你会发现：

• Claude Code 的 harness 就是一个更复杂的 ReAct Loop
• Claude Code 的 tools/ 目录就是本地工具层
• Claude Code 的 .mcp.json 配置就是这个项目的 mcp_servers.json 的增强版
• Claude Code 的 Skills 本质上也是"注入提示词"这个模式

---

2. Agent 框架全景：Harness 架构深度解析

2.1 整体架构图

┌──────────────────────────────────────────────────────────────────┐
│                        入口层 (Entry Point)                        │
│   main.py (CLI)  /  ui/app.py (Web)                              │
│   负责：接收用户输入、展示执行过程、输出最终结果                      │
└─────────────────────────────┬────────────────────────────────────┘
                              │ 传入: user_input
                              ▼
┌──────────────────────────────────────────────────────────────────┐
│                   Agent Core (Harness 主循环)                      │
│   agent/core.py                                                   │
│                                                                    │
│   while iteration  str:
        # 第0步：组装初始消息
        self.messages = [
            {"role": "system", "content": SYSTEM_PROMPT},   # 你是谁，能干什么
            {"role": "user",   "content": user_input},       # 用户想干什么
        ]

        # 第1步：进入循环 —— 这就是 Harness 的核心
        for iteration in range(MAX_TOOL_CALLS):  # 最多迭代 N 次，防止死循环

            # 1a. 把当前所有消息 + 所有可用工具定义发给 LLM
            response = self.llm.chat(
                messages=self.messages,
                tools=self.tool_schemas   # ← 所有工具的 schema 列表
            )

            # 1b. LLM 返回 assistant 消息，追加到历史
            assistant_msg = self.llm.assistant_message(response)
            self.messages.append(assistant_msg)

            # 1c. 检查 LLM 是否想调用工具
            tool_calls = self.llm.parse_tool_calls(response)

            if not tool_calls:
                # LLM 不再需要工具 = 完成了！直接输出文本
                return assistant_msg["content"]

            # 1d. LLM 想调用工具 → 逐个执行
            for tc in tool_calls:
                result = self.tool_executor(tc["name"], tc["arguments"])

                # 1e. 把工具执行结果追加到消息历史
                self.messages.append({
                    "role": "tool",
                    "tool_call_id": tc["id"],
                    "content": result,
                })

            # 1f. 回到循环顶部，LLM 看到工具结果后继续决策
            #     可能继续调工具，也可能输出最终答案

        return "达到最大迭代次数"

关键设计决策（对比 Claude Code）：

| 设计点 | 本项目 | Claude Code 的做法 | |--------|--------|-------------------| | 停止条件 | MAX_TOOL_CALLS 硬上限 | 用户可中断 + 预算管理 + 自然停止 | | 消息管理 | 全量 messages 数组 | 更复杂的上下文窗口管理（压缩、摘要） | | 工具执行 | 串行逐个执行 | 部分工具可并行执行 | | 错误处理 | 异常被捕获，注入错误消息 | 分级错误：重试/跳过/终止 | | 流式输出 | 可选 | 深度集成，实时展示工具调用过程 |

2.3 消息流转全链路

以下是一次完整调用的消息演变过程：

# ===== 初始状态 =====
messages = [
    {"role": "system", "content": "你是一个知识教学智能助手..."},
    {"role": "user",   "content": "我想学Python装饰器"},
]

# ===== 第1轮：LLM 决策 → 调用 web_search =====
# LLM 返回:
assistant_msg = {
    "role": "assistant",
    "content": None,
    "tool_calls": [
        {"id": "call_001", "function": {"name": "web_search",
         "arguments": '{"query":"Python装饰器教程入门"}'}}
    ]
}
# 执行 web_search → 返回 JSON 搜索结果
# 追加 tool 消息:
messages.append({
    "role": "tool",
    "tool_call_id": "call_001",
    "content": '{"results":[{"title":"Python装饰器详解",...}]}'
})

# ===== 第2轮：LLM 看到搜索结果，决定抓取内容 =====
# LLM 返回: tool_calls → fetch_webpage(url)
# 执行 → 追加 tool 消息

# ===== 第3轮：LLM 看到正文，决定加载编撰技能 =====
# LLM 返回: tool_calls → load_skill(name="knowledge-compiler")
# load_skill 执行 → 返回完整的 SKILL.md 正文（Layer 2 渐进式披露）
# 追加 tool 消息（内容是编撰指导）

# ===== 第4轮：LLM 读到技能指导，按格式生成第1章教材（text 响应）=====
# LLM 可能继续生成第2章、第3章...

# ===== 第N-2轮：LLM 决定加载出题技能 =====
# LLM 返回: tool_calls → load_skill(name="exercise-generator")
# 返回出题指导 → LLM 按格式为每章生成练习题

# ===== 第N轮：全部完成 =====
# LLM 返回: {"role": "assistant", "content": "我已经为你生成了完整的教材..."}
# tool_calls 为空 → 循环终止，返回 content

2.4 Harness 核心：Tool Registry 模式

在 Claude Code 的源码中，你会看到类似的模式。核心思想是：工具的定义（schema）和执行（handler）分离，通过注册表管理。

# ===== 模式抽象 =====
# 任何 Agent 框架都可以用这个模型来描述：

TOOL_REGISTRY = {
    "tool_name_1": {
        "schema": {           # 给 LLM 看的：工具签名（名称、参数、描述）
            "type": "function",
            "function": {
                "name": "...",
                "description": "...",
                "parameters": {...}
            }
        },
        "handler": callable,  # 实际执行的：Python 函数 / Skill 提示词 / MCP 远程调用
    },
    "tool_name_2": { ... },
}

# Agent 启动时：
#   1. 遍历 TOOL_REGISTRY，收集所有 schema 发给 LLM
#   2. LLM 调用某个工具时，查找对应的 handler 执行
#   3. 结果追加回消息历史

本项目把这个模式扩展为三层，每层的 handler 不同：

| 层 | handler 是什么 | schema 来源 | |----|---------------|-------------| | 本地工具 | Python 函数直接调用 | 代码中硬编码的 schema 字典 | | Skills (load_skill) | SkillLoader 返回 SKILL.md 正文 | LLM 通过 load_skill 工具按需获取（渐进式披露） | | MCP | 通过 JSON-RPC 转发到外部进程 | tools/list 动态获取，运行时转换 |

---

3. 三层能力体系的定位与关系

一个常见误区是把这三者混为一谈。它们的本质区别：

Function Call (机制)
    │
    │  是 "LLM 怎么知道要用什么工具" 的机制
    │  本质: OpenAI 原生的 tool_choice="auto" + tools 参数
    │  你的角色: 写 schema 定义，LLM 自动匹配
    │
    ├── 本地工具 (Local Tools)
    │   │
    │   │  是 "工具怎么执行" 的最直接实现
    │   │  本质: Python 函数，输入参数，返回字符串
    │   │  适用: 需要执行代码逻辑的场景（发 HTTP 请求、操作文件、计算）
    │   │  类比 Claude Code: BashTool、ReadTool、WriteTool 等
    │   │
    ├── Skills (技能)
    │   │
    │   │  是 "怎么让 LLM 高质量完成复杂任务" 的提示词工程手段
    │   │  本质: 不是执行代码，而是通过「渐进式披露」让 LLM 获取详细任务指导
    │   │        Layer 1 — system prompt 仅注入技能名 + 一行描述（~20 token/技能）
    │   │        Layer 2 — LLM 按需调用 load_skill(name) 获取完整的 SKILL.md 正文
    │   │  适用: 编撰文档、出题、翻译、审校等纯 LLM 能力可完成的任务
    │   │  类比 Claude Code: 类似 Skills 目录下的 SKILL.md 文件
    │   │
    └── MCP (外部工具)
        │
        │  是 "怎么调用别人写的工具" 的互操作协议
        │  本质: 不是你写的代码，而是别人以标准协议暴露的服务
        │  适用: 需要集成第三方能力（搜索引擎、数据库、API）
        │  类比 Claude Code: .mcp.json 配置的 MCP Servers
        │

数据流对比（同一个"获取网页内容"需求，三种实现路径）：

# 路径1: 本地工具 —— 你写的代码
LLM 调用: fetch_webpage(url="https://...")
  → tools/web_fetch.py: requests.get(url) → BeautifulSoup → 清洗 → 返回文本

# 路径2: Skill —— LLM 能力（不适用于这个需求，Skill 不能发 HTTP）
# Skills 用于纯 LLM 任务，这里仅作对比

# 路径3: MCP —— 调别人的工具
LLM 调用: mcp__fetch(url="https://...")
  → mcp/client.py: JSON-RPC tools/call → MCP Server 进程 → 返回结果

---

4. 教学：如何添加自己的 Tool

模式总结

添加一个本地工具，只需要做 3 件事：

1. 写一个 Python 函数（输入 dict → 返回 str）
2. 写一个 schema 字典（告诉 LLM 这个工具是干什么的）
3. 在 LOCAL_TOOLS 和 LOCAL_TOOL_SCHEMAS 中注册

伪例 1：添加"翻译工具"

# ===== 步骤1: 在 tools/ 下新建 translate.py =====

import json

# 1. 写函数：接收参数，返回字符串
def translate_text(text: str, target_language: str = "中文") -> str:
    """翻译工具（这里用伪逻辑演示结构，实际可接任何翻译 API）"""
    # 实际使用时，你可以接 Google Translate API 或调用 LLM
    translated = some_translate_api(text, target_language)
    return json.dumps({
        "original": text,
        "translated": translated,
        "target_language": target_language,
    }, ensure_ascii=False)

# 2. 写 schema：这是 LLM 看到的东西，描述要写清楚
TRANSLATE_SCHEMA = {
    "type": "function",
    "function": {
        "name": "translate_text",
        "description": (
            "将文本翻译成指定语言。"
            "当你需要把英文资料翻译成中文让用户阅读时使用。"
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "text": {
                    "type": "string",
                    "description": "要翻译的文本内容",
                },
                "target_language": {
                    "type": "string",
                    "description": "目标语言，如'中文'、'英文'、'日语'",
                },
            },
            "required": ["text", "target_language"],
        },
    },
}

# ===== 步骤2: 在 tools/executor.py 中注册 =====

# 导入
from tools.translate import translate_text, TRANSLATE_SCHEMA

# 注册函数映射
LOCAL_TOOLS["translate_text"] = translate_text

# 注册 schema
LOCAL_TOOL_SCHEMAS.append(TRANSLATE_SCHEMA)

# 完成！LLM 现在可以调用 translate_text 了

伪例 2：添加"计算器工具"

# tools/calculator.py

def calculate(expression: str) -> str:
    """安全地计算数学表达式"""
    import json
    try:
        # 安全计算（仅允许数学运算，禁止危险函数）
        allowed = {"__builtins__": {}}
        result = eval(expression, allowed, {"__builtins__": {}})
        return json.dumps({"expression": expression, "result": result})
    except Exception as e:
        return json.dumps({"error": str(e)})

CALCULATOR_SCHEMA = {
    "type": "function",
    "function": {
        "name": "calculate",
        "description": "计算数学表达式，如 '2 + 3 * 4'",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "数学表达式",
                },
            },
            "required": ["expression"],
        },
    },
}

要点总结

| 关注点 | 注意事项 | |--------|---------| | 函数签名 | 参数必须与 schema 中 properties 的 key 完全匹配 | | 返回值 | 必须是 str（通常用 json.dumps 包装） | | Schema 描述 | 这是 LLM 唯一能看到的信息，描述不清楚 LLM 就不会正确调用 | | required | 必须列出所有必填参数 | | 错误处理 | 不要抛异常，捕获后用 json.dumps({"error": "..."}) 返回 |

---

伪例 3：TodoWrite 任务管理工具（已内置本项目）

这是本项目已实现的一个重要工具，展示了"工具不止是执行逻辑，还可以和框架层双向交互"的设计模式。

问题背景： 在多步复杂任务中，LLM 容易"迷路"——忘记做过什么、重复执行已完成步骤、遗漏子任务。TodoWrite 通过两个机制解决这个问题：

机制 1: 强制顺序聚焦
  → 同时只能有一个 in_progress 任务
  → 强制性要求 LLM 做完一个再做下一个

机制 2: 静止检测 + 提醒注入
  → 框架层计数：连续 N 轮不调用 todo_write
  → 超过阈值 (3轮) → 自动在消息列表中注入提醒
  → 提醒内容包括当前任务状态，引导 LLM 回到正轨

核心代码解析：

# tools/todo_write.py 的核心逻辑

class TodoManager:
    def __init__(self):
        self.items: list[dict] = []       # 当前任务列表
        self._rounds_without_todo = 0     # 未调用 todo 的连续轮次

    def update(self, items: list) -> str:
        """验证并更新任务列表，强制只有一个 in_progress"""
        validated, in_progress_count = [], 0
        for item in items:
            status = item.get("status", "pending")
            if status == "in_progress":
                in_progress_count += 1
            validated.append({
                "id": item["id"], "text": item["text"], "status": status
            })
        if in_progress_count > 1:
            raise ValueError("Only one task can be in_progress")
        self.items = validated
        self._rounds_without_todo = 0     # 调用了 todo → 重置计数
        return self.render()

    def mark_round(self, had_todo_call: bool) -> str | None:
        """每轮 Agent 循环结束后调用，计数并决定是否注入提醒"""
        if had_todo_call:
            self._rounds_without_todo = 0
            return None
        self._rounds_without_todo += 1
        if self._rounds_without_todo >= REMINDER_THRESHOLD:  # 3轮
            return self._build_reminder()
        return None

框架层注入逻辑：

# agent/core.py 中的集成

class Agent:
    def run(self, user_input: str) -> str:
        for iteration in range(MAX_TOOL_CALLS):
            # 第1步: 每轮开始先检查是否需要注入提醒
            reminder = mark_todo_round(False)  # 先假设本轮无 todo 调用
            if reminder:
                # 作为 system 消息注入到对话历史
                self.messages.append(
                    {"role": "system", "content": f"[TodoWrite 提醒] {reminder}"}
                )

            # 第2步: LLM 决策
            response = self.llm.chat(...)
            tool_calls = parse_tool_calls(response)

            # 第3步: 本轮结束后，告知 TodoManager 是否调用了 todo_write
            had_todo = any(tc["name"] == "todo_write" for tc in tool_calls)
            mark_todo_round(had_todo)  # 如果调用了 → 重置计数器

完整调用时序：

轮次 1: LLM 调用 todo_write(规划任务)  → had_todo=True  → 计数器重置为 0
轮次 2: LLM 调用 web_search            → had_todo=False → 计数器 = 1
轮次 3: LLM 调用 fetch_webpage         → had_todo=False → 计数器 = 2
轮次 4: LLM 调用 fetch_webpage         → had_todo=False → 计数器 = 3 → 注入提醒！
  → 注入: "[TodoWrite 提醒] 已连续3轮未更新任务列表。请调用 todo_write 检查进度..."
轮次 5: LLM 收到提醒，调用 todo_write(标记完成+开启新任务) → 计数器重置

设计要点：

| 要点 | 说明 | |------|------| | 双向交互 | Tool 不再只是"被调用→返回结果"，而是通过全局状态和框架层互相感知 | | 计数位置 | mark_round(False) 在每轮 开始时 调用（注入提醒），mark_round(had_todo) 在每轮 结束后 调用（更新计数器） | | 提醒内容 | 如果任务列表非空，提醒会包含当前所有任务的渲染结果，帮助 LLM 快速回忆上下文 | | 合并模式 | merge=true 支持增量添加新任务而不丢失已有进度 |

---

伪例 4：Subagent 子任务拆分工具（已内置本项目）

这是本项目已实现的另一个重要工具，展示了"消息隔离 + 分治法"模式。

问题背景： Agent 工作越久，messages 数组越臃肿——每一轮的工具调用和结果都被追加到消息历史中。在长任务中，这会导致：

• context window 被大量工具调用记录占满
• LLM 容易被过长的历史分散注意力
• 不同子任务的信息互相污染

Subagent 的解决方案：大任务拆小，message 分离，每个子任务用全新上下文执行。

主 Agent (Parent)                    子 Agent 1 (Child)              子 Agent 2 (Child)
┌────────────────────┐              ┌──────────────────┐            ┌──────────────────┐
│ messages:          │              │ messages:        │            │ messages:        │
│ [system, user,     │              │ [sub_system,     │            │ [sub_system,     │
│  assistant, tool,  │   task()     │  user_prompt,    │   task()   │  user_prompt,    │
│  assistant, tool,  │──────────────│  assistant,      │────────────│  assistant,      │
│  ...数十轮...]     │  返回摘要     │  tool, ...]      │  返回摘要   │  tool, ...]      │
│                    │◄─────────────│                  │◄───────────│                  │
│                    │              │  ← 上下文隔离 →   │            │  ← 上下文隔离 →   │
└────────────────────┘              └──────────────────┘            └──────────────────┘

工具分层设计：

# 主 Agent 工具集 = 子 Agent 工具集 + task 工具
PARENT_TOOLS = CHILD_TOOLS + [task]
# 子 Agent 工具集 = 所有本地工具（不含 task，防止递归爆炸）
CHILD_TOOLS = {web_search, fetch_webpage, save_document, load_skill, todo_write}

核心代码解析：

# tools/subagent.py 的核心逻辑

from agent.llm_client import LLMClient
from config import SUBAGENT_MAX_TOOL_CALLS, build_subagent_system_prompt

def run_subagent(prompt: str) -> str:
    """主 Agent 调用此函数 spawn 一个子 Agent"""
    llm = LLMClient()
    system_prompt = build_subagent_system_prompt()

    # 全新的 messages 数组 —— 上下文隔离的关键！
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": prompt},   # 只有主 Agent 传过来的 prompt
    ]

    # 子 Agent 自己的 ReAct 循环
    for iteration in range(SUBAGENT_MAX_TOOL_CALLS):
        response = llm.chat(messages=messages, tools=CHILD_TOOL_SCHEMAS)
        assistant_msg = llm.assistant_message(response)

        if assistant_msg is None:
            break

        messages.append(assistant_msg)

        tool_calls = llm.parse_tool_calls(response)
        if not tool_calls:
            # 子 Agent 完成 → 返回文本摘要给主 Agent
            return assistant_msg.get("content", "")

        # 执行子 Agent 的工具调用
        for tc in tool_calls:
            handler = CHILD_TOOLS.get(tc["name"])
            result = handler(**tc["arguments"])
            messages.append(llm.format_tool_result(tc["id"], result))

    return "(Subagent 已达到最大工具调用次数)"

完整调用时序：

主 Agent 轮次 1-3: 搜索、抓取、分析... messages 已累积 12 条

主 Agent 轮次 4: 决定分治法 → 调用 task(prompt="搜索微积分入门资料并返回3篇最佳文章摘要")
  ├── run_subagent() 被调用
  │   ├── sub_messages = [sub_system, user_prompt]  ← 全新的 2 条消息
  │   ├── 子轮1: web_search("微积分入门教程")        ← sub_messages: 4 条
  │   ├── 子轮2: fetch_webpag

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [Ghostricke9](https://github.com/Ghostricke9)
- **Source:** [Ghostricke9/StudyAgent](https://github.com/Ghostricke9/StudyAgent)
- **License:** MIT

Install and usage instructions live in the source repository linked above.