ArchiveBox MCP 服务器源码解析:基于 Click 动态反射的 AI Agent 工具层
2026/9/21 19:04:38 网站建设 项目流程
  • 后端
  • 数据工程

【免费下载链接】ArchiveBox

🗃 Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...

项目地址:https://gitcode.com/gh_mirrors/ar/ArchiveBox
点击查看免费下载

导读:本文围绕 ArchiveBox 的 Model Context Protocol(MCP)服务器实现(archivebox/mcp/server.py)展开,深入剖析它如何通过动态反射现有 Click CLI 命令树,将 ArchiveBox 的归档、搜索、抓取等能力自动暴露为可供 AI Agent 发现与调用的 MCP 工具。读完本文,你将掌握该模块的协议设计、工具发现机制、参数类型映射、JSON-RPC 请求分发与 stdio 运行方式,并能在本地直接驱动archivebox mcp进行验证。

模块定位:一个面向 AI Agent 的轻量无状态网关

archivebox.mcp.server是 ArchiveBox 中 Model Context Protocol 服务器的核心实现。它的设计目标非常明确:把 ArchiveBox 已有的 Click CLI 命令自动包装成 MCP 工具,让 AI Agent(如 Claude、各类 MCP 客户端)能够通过标准协议发现并执行归档操作,而不需要为每个命令手写一份工具定义。

其设计要点(可在 archivebox/mcp/README.md 的 Features 一节确认)包括:

  • 自动发现:动态遍历 Click CLI 命令树,自动生成工具定义,无需手工维护 Schema;
  • 零重复:完全复用现有 Click 命令的类型、参数与帮助文本;
  • 自动同步:CLI 命令变更后,MCP 工具定义随之变化;
  • 无状态:不引入数据库模型或状态管理;
  • 轻量:核心代码约 350 行(见 archivebox/mcp/server.py)。

从架构上看,MCP 应用(Django app)仅承担模块注册职责(archivebox/mcp/apps.py 定义MCPConfigverbose_name为 "Model Context Protocol Server"),真正的协议逻辑全部收敛在server.py一个文件中,这也是 apidocs 中该模块文档(docs/apidocs/archivebox/archivebox.mcp.server.md)所列举的全部内容。

对外暴露的工具面:六个聚焦的 Agent 友好工作流

server.py顶部用三个常量定义了工具面的边界(第 18-22 行):

PROTOCOL_VERSION = "2025-11-25" PUBLIC_TOOLS = ("add", "search", "crawl", "snapshot", "archiveresult", "shell") ACTION_TOOLS = {"crawl", "snapshot", "archiveresult"} READ_ONLY_ACTIONS = {"help", "list", "search", "status", "version"} DESTRUCTIVE_ACTIONS = {"delete", "remove"}

其中PROTOCOL_VERSION声明遵循 MCP 2025-11-25 规范。尽管 ArchiveBox 的 CLI 拥有helpversioninitinstallupdateconfigscheduleservermanage等大量子命令(完整清单见 archivebox/cli/init.py 中ArchiveBoxGroupall_subcommands),MCP 服务器只暴露六个精挑细选的工作流工具:

工具名类型说明
add单命令工具添加并归档 URL(走 JSONL/标准输入管线)
search单命令工具深度搜索归档内容(只读)
crawl动作组工具action=create/list/update/delete管理 Crawl 记录
snapshot动作组工具action管理 Snapshot 记录
archiveresult动作组工具action管理 ArchiveResult 记录
shellPython 逃生舱在已初始化的 ArchiveBox Django shell 中执行任意 Python

这一设计与测试用例 archivebox/tests/test_cli_mcp.py 中test_mcp_exposes_six_focused_tools的断言完全一致:set(tools_by_name) == {"add", "search", "crawl", "snapshot", "archiveresult", "shell"},且len(tools) == 6

为什么要收窄工具面?因为crawlsnapshotarchiveresult是 ArchiveBox 的三类核心记录模型(CRUD 语义统一),将其包装为带action选择器的动作组工具,可以显著减少 Agent 需要记忆的工具数量,降低调用复杂度。

动态工具发现:从 Click 命令树到 MCP 工具定义

MCP 服务器不维护任何硬编码工具清单,而是通过递归遍历 Click 命令树完成"零 Schema 定义"的自动发现。

工具载体:MCPTool

@dataclass(frozen=True) class MCPTool: """A discovered leaf Click command and its ArchiveBox command path.""" name: str command_path: tuple[str, ...] command: click.Command

MCPTool用一个不可变 dataclass 绑定"工具名 + 命令路径 + 原始 Click 命令对象"。其中工具名由命令路径拼接而成,_discover_commandtool_name = "_".join(command_path).replace("-", "_"),例如archivebox crawl create路径生成工具名crawl_create

递归发现流程

MCPServer.get_tools()(第 418-429 行)以ArchiveBoxGroup(来自 archivebox/cli/init.py,一个 lazy-loading 的 Click Group)为根,只对PUBLIC_TOOLS中的六个命令执行_discover_command

  • 若当前命令是click.Group,则遍历list_commands(ctx)递归下钻;
  • 若是叶子命令,则登记为一个MCPTool

由于ArchiveBoxGroup.get_command采用按需惰性导入(_lazy_load),每个子命令模块只有在真正被访问时才加载,因此工具发现过程的开销可控,并且每次启动都能拿到与当前安装完全一致的 CLI 元数据——这正是"CLI 变更自动同步到 MCP 工具"的实现基础。

单命令工具与动作组工具的分流

get_public_tool_definitions()(第 431-448 行)决定每个工具如何呈现:

  • addsearch:直接通过click_command_to_mcp_tool转换为单命令工具;
  • crawlsnapshotarchiveresult:先收集所有以该组为根的叶子工具,再通过click_group_to_mcp_tool合并为一个带action枚举选择器的动作组工具;
  • shell:走专门的shell_to_mcp_tool

动作组工具的描述中会为每个 action 生成一行帮助(- {action}: {command help}),并把所有子命令的参数 Schema 合并进同一个properties。若同名参数在不同 action 下的默认值不一致,则删除该参数的default字段以免误导 Agent(第 207-211 行)。

参数类型映射:Click 类型到 JSON Schema 的自动转换

这是"零手写 Schema"的关键一环。click_type_to_json_schema_type(click_type)(第 49-74 行)把 Click 的参数类型逐一映射为 JSON Schema:

Click 类型JSON Schema
StringParamType{"type": "string"}
IntParamType{"type": "integer"}
FloatParamType{"type": "number"}
BoolParamType{"type": "boolean"}
Choice{"enum": [...choices], "type": "string"}(若全为字符串)
Path/File{"type": "string", "description": "File or directory path"}
Tuple{"type": "array", "prefixItems": [...], "minItems": N, "maxItems": N}
其他兜底{"type": "string"}

click_command_to_mcp_tool(第 108-170 行)随后遍历command.params,为每个参数补充description(取自 Click 的help)、default,处理multiple/nargs != 1的多值参数(包装为数组),并将必填参数收集进required列表。最终生成的工具定义包含nametitledescription(自动追加Equivalent CLI command: archivebox ...提示)、inputSchemaoutputSchemaannotations

一个值得注意的细节:如果命令的帮助文本中出现 "stdin" 字样(command_accepts_stdin,第 77-89 行),工具会自动附加两个参数——records(JSONL 记录数组)和stdin(原始文本),让 Agent 可以直接传递结构化记录而无需自行拼装管道。测试test_mcp_crawl_create_returns_structured_json正是通过crawlrecords/urls参数完成调用并断言返回了结构化 JSON。

参数反序列化:从 MCP JSON 参数到 Click argv

工具调用方向上的核心是arguments_to_cli(第 280-311 行),它完成 MCP JSON 参数到 Click argv(以及可选 stdin)的逆转换:

  1. 从参数中剥离recordsstdin;两者同时出现时报ValueError;仅提供records时逐条json.dumps拼成 JSONL 文本作为 stdin;
  2. 校验未知参数:sorted(set(supplied) - set(param_map))中任何未知键都会抛出Unknown argument(s): ...
  3. 按参数类型分流:click.Argument追加为位置参数;click.Option交给_option_args序列化。

_option_args(第 261-277 行)是序列化的精细部分:它优先使用选项的--xxx拼写;布尔标志(is_bool_flag)在值为真时输出主选项、为假且有secondary_opts时输出反向选项(如--no-xxx);multiple选项则逐个展开为--opt value1 --opt value2形式。

命令执行:基于 CliRunner 的进程内调用

execute_click_command(第 333-362 行)是真正的执行入口:

from archivebox.cli import cli result = CliRunner().invoke( cli, [*tool.command_path, *cli_args], input=stdin_text, prog_name="archivebox", catch_exceptions=False, )

它没有启动子进程,而是用 Click 内置的CliRunner当前 Python 进程内调用整个archiveboxCLI 入口。这意味着:

  • 工具调用天然共享进程内的 Django 初始化与配置上下文,无需重复加载;
  • 捕获到click.ClickExceptionclick.UsageErrorOSErrorSystemExitValueError时,会打印 traceback 并返回exitCode=1的结构化错误。

统一输出 envelope

无论成功失败,_tool_result(第 364-388 行)都会返回同一套结构化包络:

{ "command": "archivebox crawl create", "success": bool, "error": str | None, "exitCode": int, "records": [...], # 从 stdout 解析出的结构化记录 "stdout": str, "stderr": str, }

该包络的 JSON Schema 由tool_output_schema()(第 173-188 行)统一定义,作为所有工具的outputSchema。同时,结果会以content[0].text(JSON 文本)与structuredContent(结构化对象)两种形式返回——前者兼容只支持文本内容的 MCP 客户端,后者让理解 JSON 的 Agent 直接消费结构化数据。测试test_mcp_crawl_create_returns_structured_json断言json.loads(result["content"][0]["text"]) == result["structuredContent"],验证了两者的严格一致。

records的解析由parse_structured_records(第 314-330 行)完成:优先整体json.loads为数组;失败则逐行按 JSONL 解析;仍失败则返回空列表,绝不猜测人类可读输出。

MCPJSONEncoder:CLI 值的 JSON 兼容化

CLI 输出中常混有 UUID、Path、Click 哨兵对象(click.core._SentinelClass)等无法直接序列化的值。MCPJSONEncoder(第 25-37 行)在default方法中统一处理:哨兵值输出为None、元组转列表、其余兜底str()。所有json.dumps(请求解析后的响应、stdin 拼装、结果编码)都使用该编码器。

JSON-RPC 2.0 请求分发:MCPServer 的方法路由

MCPServer.handle_request(第 502-542 行)实现 JSON-RPC 2.0 的方法分发。它首先识别通知(无id字段的请求直接返回None,不产生响应),然后按方法名路由:

方法处理函数说明
initializehandle_initialize握手,返回协议版本、能力声明与服务信息
ping内联返回空{}
tools/listhandle_tools_list返回六个公开工具定义
tools/callhandle_tools_call按工具名分发执行
其他内联返回-32601 Method not found

错误码遵循 JSON-RPC 规范:参数错误(TypeError/ValueError)返回-32602 Invalid params;执行期 Click/OS/运行时错误返回-32603 Internal error(附 traceback 作为data);JSON 解析失败在 stdio 层返回-32700 Parse error

initialize:握手信息

handle_initialize(第 450-465 行)返回protocolVersion: "2025-11-25"capabilities.tools.listChanged: False,以及serverInfo(名称archivebox-mcp、标题ArchiveBox、版本取 archivebox/config/version.py 中的VERSION,该值通过 pip 元数据或pyproject.toml自动探测)。它还携带一段instructions文本,指导 Agent 何时使用哪个工具、如何把返回的 records 直接回传给 update/delete 动作,以及仅在 curated 工具无法表达操作时才使用shell

tools/call:动作组与 shell 的特殊分流

handle_tools_call(第 470-500 行)除了校验工具名与参数类型外,还处理两类特殊情况:

  • shell:只接受code参数(非空字符串),并将其重写为{"args": ["--plain", "--quiet-load", "-c", code]},即等价于archivebox shell --plain --quiet-load -c CODE
  • 动作组:弹出action参数并校验其属于该组的可用动作(如crawlcreate/delete/list/update),再把工具名改写为crawl_create之类的具体叶子工具。非法 action 会得到-32602协议错误,错误消息列出可选动作——测试test_mcp_invalid_action_is_a_protocol_error验证了这一点。

安全注解:向 Agent 描述命令副作用

MCP 2025-11-25 规范支持工具注解,tool_annotations(第 92-105 行)根据命令路径自动推导:

  • readOnlyHint:action 属于{help, list, search, status, version}时为True
  • destructiveHint:action 属于{delete, remove}时为True
  • idempotentHint:只读或破坏性命令均为True(重复执行无累积副作用);
  • openWorldHint:根命令为add/extract/oneshot/run/update,或crawl/snapshotcreate动作时为True(表示会访问外部网络世界)。

shell工具被显式标记为readOnlyHint=FalsedestructiveHint=TrueopenWorldHint=False,并在描述中警告它"拥有对集合数据库与文件系统的完整访问权限"(shell_to_mcp_tool,第 228-258 行)。这相当于给 Agent 一个明确的信任边界提示:只有 curated 工具无法表达的操作才应动用 Python 逃生舱。

stdio 服务循环与进程入口

run_stdio_server:逐行 JSON-RPC

run_stdio_server(第 544-561 行)是服务器的运行核心,以"一行一个 UTF-8 JSON 消息"的方式与宿主进程通信:

for line in sys.stdin: if not line.strip(): continue try: request = json.loads(line) response = self.handle_request(request) if response is not None: print(json.dumps(response, cls=MCPJSONEncoder), flush=True) except json.JSONDecodeError as err: # 返回 -32700 Parse error

空行被跳过、通知请求不产生响应、每条响应通过flush=True立即写出,避免与 MCP 客户端(Claude Desktop、编辑器 MCP 插件等)交互时出现缓冲延迟。

run_mcp_server:模块级入口

def run_mcp_server() -> None: """Start the ArchiveBox MCP stdio server.""" MCPServer().run_stdio_server()

run_mcp_server是模块暴露的顶层函数,也是 apidocs 文档中最后一个函数条目。它被 CLI 入口调用。

CLI 入口与实战用法

MCP 服务器通过archivebox mcp子命令启动,定义在 archivebox/cli/archivebox_mcp.py(注册于ArchiveBoxGroup.meta_commands,见 archivebox/cli/init.py 第 32-37 行):

archivebox mcp < requests.jsonl > responses.jsonl

交互式用法(initializetools/list):

archivebox mcp {"jsonrpc":"2.0","id":1,"method":"initialize","params":{}} {"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}

一次完整的tools/call请求示例:

{ "jsonrpc":"2.0", "id":3, "method":"tools/call", "params":{ "name":"crawl", "arguments":{"action":"create", "urls":["https://example.com/"], "depth":1} } }

Python 客户端最小示例(来自 archivebox/mcp/README.md):

import json import subprocess request = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}} completed = subprocess.run( ["archivebox", "mcp"], input=json.dumps(request) + "\n", capture_output=True, text=True, check=True, timeout=30, ) response = json.loads(completed.stdout) assert response["id"] == 1 assert response["result"]["serverInfo"]["name"] == "archivebox-mcp"

也可以直接在 Python 中调用MCPServer().handle_request(...)handle_tools_call(...),无需经过 stdio——这在集成测试与二次开发中非常方便(参考 archivebox/tests/test_cli_mcp.py 的用法)。

测试验证:行为即规范

archivebox/tests/test_cli_mcp.py 完整覆盖了模块的关键行为,可作为理解实现的活文档:

  • test_mcp_stdio_handles_handshake_notification_and_ping:验证 stdio 模式下的握手、通知(无响应)与 ping;
  • test_mcp_exposes_six_focused_tools:验证六个工具、crawlaction枚举、search的只读注解、shell的必填code参数;
  • test_mcp_crawl_create_returns_structured_json:端到端验证crawl create的结构化返回;
  • test_mcp_snapshot_update_accepts_records_without_a_jsonl_pipeline:验证"把上一个动作返回的 records 直接传给 update"的 Agent 工作流;
  • test_mcp_cli_errors_are_structured_for_agents:验证 CLI 层错误(如缺 URL)被封装为success=false+exitCode=1的结构化结果;
  • test_mcp_invalid_action_is_a_protocol_error:验证非法 action 返回-32602协议错误;
  • test_mcp_shell_runs_python_through_archivebox_shell:验证shell工具在初始化后的 Django 环境中执行 Python 并回传 stdout。

小结

archivebox.mcp.server的价值在于一种"CLI 即 API"的工程思路:以 Click 元数据为唯一事实来源,用不到 400 行代码构建出一个零手工 Schema、自同步、可测试的 MCP 服务器。无论你是在研究 MCP 协议的实现细节,还是想把自家 Click/Argparse 工具快速暴露给 AI Agent,这个模块都是一个值得参照的紧凑范本。进一步阅读建议:模块文档 docs/apidocs/archivebox/archivebox.mcp.server.md、模块说明 archivebox/mcp/README.md、核心实现 archivebox/mcp/server.py 与测试 archivebox/tests/test_cli_mcp.py。

  • 后端
  • 数据工程

【免费下载链接】ArchiveBox

🗃 Open source self-hosted web archiving. Takes URLs/browser history/bookmarks/Pocket/Pinboard/etc., saves HTML, JS, PDFs, media, and more...

项目地址:https://gitcode.com/gh_mirrors/ar/ArchiveBox
点击查看免费下载
上一篇:Ip2region:高效离线IP地址定位库
下一篇:FreeMove 开源项目使用教程

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询