1. 从“工具海”到“工具池”一个被忽视的WebMCP核心范式如果你最近在折腾AI Agent或者LLM应用开发大概率听说过WebMCPModel Context Protocol这个协议。它被设计为连接LLM与外部工具和数据的桥梁听起来很美好但上手后很多人会陷入一个困境为了一个功能丰富的Agent我是不是得预先注册几十上百个工具配置文件越来越臃肿启动越来越慢维护成本直线上升这和我们追求的灵活、智能的Agent似乎背道而驰。这正是我最初接触WebMCP时的困惑。直到我在CreatorWeave这个项目中实践了“On-Demand”按需工具注册模式才真正理解了WebMCP设计哲学中一个被严重低估的侧面。它的精髓不在于你能注册多少工具而在于你能否用最少的、最核心的工具去动态地、智能地“代理”出无限可能的功能。这篇文章我就来拆解这个“只注册2个工具代理N个”的实践它不仅仅是技术实现更是一种架构思维的转变。简单来说我们不再把Agent看作一个拥有固定技能列表的“瑞士军刀”而是把它看作一个拥有核心“元能力”的“指挥官”。这个指挥官手里只有两把万能钥匙两个核心注册工具但它知道如何根据任务指令动态地调用、组合外部资源即被代理的N个工具或服务来完成任务。这极大地提升了Agent的灵活性、可维护性和响应速度。无论你是想构建一个能处理复杂工作流的创作助手还是一个能连接企业内部各种API的业务Agent这个思路都能帮你跳出“工具注册地狱”。2. 为什么是“On-Demand”重新审视WebMCP的工具管理在深入代码之前我们必须先厘清一个根本问题为什么传统的“预注册所有工具”模式在复杂场景下会失效以及WebMCP协议本身是否支持更动态的方式2.1 传统预注册模式的三大痛点第一启动与内存开销。一个WebMCP Server在启动时需要加载并初始化所有注册的工具函数、它们的schemaJSON Schema描述。如果工具数量庞大例如超过50个这个初始化过程会显著拖慢启动速度并且这些工具的定义会常驻内存即使整个会话周期一次都用不到。第二配置与维护的复杂性。mcp_server的配置文件如server.py或config.json会变得极其冗长。每增加、删除或修改一个工具都需要更新这个配置文件并重启服务。在快速迭代的开发阶段或者工具集动态变化的场景下这简直是噩梦。第三功能暴露的“噪音”。当Agent如Claude Desktop、Cursor等连接到你的Server时它会收到一个包含所有工具签名的庞大列表。这可能会干扰LLM的决策让它陷入“选择困难症”或者在不合适的时机调用不相关的工具。2.2 WebMCP协议提供的灵活性工具Tools与资源ResourcesWebMCP协议的设计其实已经考虑到了动态性。它不仅仅有Tools工具还有Resources资源。Tools代表可执行的操作函数而Resources代表可读取的数据或状态如文件内容、数据库查询结果。协议允许Server在运行时动态地公布notify新的Resources给Client。虽然协议标准更侧重于Resources的动态性但Tools的动态注册与注销在实现层面并非不可能。许多MCP Server的实现包括官方和社区的都提供了运行时更新工具列表的接口。“On-Demand”模式的核心思想就是利用这种潜力将工具的“注册”动作延迟到真正需要它的前一刻或者通过一个“代理工具”来间接调用。2.3 “On-Demand”模式的核心优势对比之下On-Demand模式的优势就非常明显了极简启动Server启动时只加载最核心、最通用的几个工具例如2个启动飞快内存占用小。动态能力Agent的能力边界不再是固定的可以根据用户请求的上下文动态地加载或指向新的功能模块。解耦与维护新增功能模块即被代理的N个工具可以独立开发、测试、部署只需确保它们能被核心的“代理工具”访问到即可无需修改主Server的配置。清晰的架构两个核心工具充当了明确的“网关”或“路由器”使得整个系统的数据流和控制流更加清晰。3. 核心设计哪两个工具足以“代理万物”这是整个模式最巧妙也最需要深思熟虑的部分。选择哪两个工具作为“万能钥匙”决定了整个架构的形态和易用性。在CreatorWeave的实践中我经过多次迭代最终确定了两个方向一个用于“执行”一个用于“发现与管理”。这里我提供两种经过验证的设计方案。3.1 方案一函数执行器 工具注册器经典网关模式这是最直接、最强大的模式类似于一个微型的“云函数”平台或“插件系统”。execute_function函数执行器职责接收一个函数名和参数在安全沙箱或特定上下文中执行对应的函数并返回结果。工作原理Server背后维护着一个工具函数注册表例如一个Python字典或数据库。这个注册表最初是空的或者只包含少数内置函数。execute_function工具被调用时它根据传入的function_name去注册表中查找对应的函数对象然后用arguments参数调用它。如何实现“代理N个”那“N个”工具的实现就是一个个独立的函数它们被预先编写好存储在一个特定的目录如./tools/或模块中。当需要某个工具时可以通过另一个工具或初始化脚本将这些函数动态地“注入”到注册表中。execute_function本身并不关心它执行的是哪个函数它只是一个安全的调用网关。# 伪代码示例execute_function 的核心逻辑 class ToolRegistry: _functions {} # 全局函数注册表 classmethod def register(cls, name: str, func: callable, schema: dict): cls._functions[name] {func: func, schema: schema} classmethod def execute(cls, name: str, arguments: dict) - str: if name not in cls._functions: return fError: Function {name} not found. tool_info cls._functions[name] try: result tool_info[func](**arguments) return str(result) except Exception as e: return fError executing {name}: {e} # WebMCP Tool 定义 async def execute_function(name: str, arguments: dict) - str: return ToolRegistry.execute(name, arguments)register_tool工具注册器职责接收一个新工具的定义包括名称、描述、参数schema和函数引用将其动态添加到ToolRegistry中。工作原理这是实现“On-Demand”的关键。Agent在运行过程中如果判断需要某个尚未加载的工具它可以先调用register_tool将工具的定义从磁盘、网络或代码字符串中加载并注册然后再调用execute_function来使用它。更高级的玩法是Agent可以分析用户需求自动生成工具的描述和参数schema然后注册一个“虚拟工具”其执行函数可能是一个对LLM的二次调用或一个复杂的工作流。# 伪代码示例register_tool 工具 async def register_tool(tool_name: str, description: str, parameters_schema: dict, source_code_or_path: str) - str: # 1. 验证schema # 2. 从source_code_or_path加载或编译函数需要安全考虑 # 3. 将函数对象和schema注册到ToolRegistry ToolRegistry.register(tool_name, loaded_function, parameters_schema) return fTool {tool_name} registered successfully.这个方案的强大之处在于它将工具的“声明”和“执行”完全分离并且将“注册”本身也工具化了。Agent获得了在运行时扩展自身能力的可能性。3.2 方案二工作流执行器 资源查询器面向流程模式如果你的N个工具通常是按照特定顺序、为了完成一个特定目标如“写博客”、“分析数据”而组合使用的那么这种模式更合适。它更侧重于编排而非单个工具的动态注册。run_workflow工作流执行器职责接收一个工作流标识符如workflow_id和输入参数执行一个预定义的工作流。工作原理工作流是一个有向无环图DAG节点是一个个具体的工具或操作即那“N个”工具边定义了执行顺序和参数传递。run_workflow工具背后连接着一个工作流引擎可以是简单的Python脚本也可以是像Prefect、Airflow这样的框架。当被调用时它启动对应的工作流并管理其中各个节点的执行。这里的“N个工具”是工作流内部的固定步骤但对WebMCP Server来说它只暴露了run_workflow这一个执行接口。query_available_workflows资源查询器职责列出当前所有可用的工作流及其描述、输入参数。工作原理它扫描工作流定义目录或者查询工作流引擎的元数据然后以结构化数据如JSON的形式返回。这相当于一个动态的“能力目录”帮助LLM了解当前可以发起哪些复杂的任务。这个方案的优势是对于复杂、多步骤的任务封装性好用户体验流畅。用户或Agent只需要说“帮我执行数据分析流程”而不需要关心内部调用了多少个API、做了多少次数据转换。劣势是工作流需要预先定义动态创建和修改工作流本身比较复杂。在CreatorWeave中我主要采用了方案一函数执行器注册器因为它提供了最大的灵活性允许Agent在对话中实时“学习”并使用新工具。接下来我们就看看如何具体实现它。4. 实战构建一个极简On-Demand WebMCP Server让我们用Python和mcpSDK快速搭建一个原型。假设我们的目标是构建一个Server它启动时只注册execute_function和register_tool。然后我们通过一个外部命令或Agent的初始请求动态加载一个get_weather获取天气工具并成功调用它。4.1 项目初始化与基础结构首先创建项目并安装依赖。mkdir on-demand-mcp-server cd on-demand-mcp-server python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install mcp创建主服务器文件server.py和工具注册表模块tool_registry.py。tool_registry.py内容如下# tool_registry.py import importlib.util import sys import json from typing import Dict, Any, Callable, Optional class ToolRegistry: 全局工具注册中心采用单例模式简化 _instance None _tools: Dict[str, Dict[str, Any]] {} # {tool_name: {func: callable, schema: dict}} def __new__(cls): if cls._instance is None: cls._instance super(ToolRegistry, cls).__new__(cls) return cls._instance def register(self, name: str, func: Callable, schema: dict): 注册一个工具到中心仓库 if name in self._tools: print(f[Warning] Tool {name} is being overwritten.) self._tools[name] {func: func, schema: schema} print(f[Info] Tool {name} registered.) def get(self, name: str) - Optional[Dict[str, Any]]: 根据名称获取工具信息 return self._tools.get(name) def list_tools(self) - Dict[str, dict]: 列出所有已注册的工具返回schema供MCP使用 return {name: info[schema] for name, info in self._tools.items()} def execute(self, name: str, arguments: dict) - str: 执行指定工具 tool_info self.get(name) if not tool_info: return json.dumps({error: fTool {name} not found in registry.}) try: # 调用实际的函数 result tool_info[func](**arguments) # 将结果转换为字符串对于复杂对象可以JSON序列化 if isinstance(result, (dict, list)): return json.dumps(result, ensure_asciiFalse) return str(result) except TypeError as e: return json.dumps({error: fInvalid arguments for {name}: {e}}) except Exception as e: return json.dumps({error: fExecution error for {name}: {e}}) # 全局注册表实例方便导入 registry ToolRegistry()4.2 实现核心的“两个工具”现在在server.py中我们基于tool_registry实现那两个核心的MCP工具。# server.py import asyncio from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import json from tool_registry import registry # 创建MCP Server实例 app Server(on-demand-mcp-server) app.list_tools() async def handle_list_tools(): 列出当前已注册的所有工具。这里只返回两个核心工具。 # 核心工具1: execute_function execute_schema { name: execute_function, description: 执行一个在注册表中已注册的工具函数。, inputSchema: { type: object, properties: { name: { type: string, description: 要执行的工具函数名称。 }, arguments: { type: object, description: 传递给工具函数的参数字典。, additionalProperties: True # 参数取决于具体工具 } }, required: [name, arguments] } } # 核心工具2: register_tool register_schema { name: register_tool, description: 动态注册一个新的工具到服务器。工具定义需包含可执行的代码或路径。, inputSchema: { type: object, properties: { tool_name: { type: string, description: 新工具的唯一名称。 }, description: { type: string, description: 工具的功能描述。 }, parameters_schema: { type: object, description: 符合JSON Schema的工具参数定义。, additionalProperties: True }, source_type: { type: string, enum: [inline_python, file_path], description: 工具源码的提供方式。 }, source_content: { type: string, description: 根据source_type可以是Python代码字符串或文件路径。 } }, required: [tool_name, description, parameters_schema, source_type, source_content] } } # 注意这里我们只返回这两个核心工具。 # 被动态注册的工具如get_weather不会在这里直接列出。 # 但LLM可以通过其他方式如对话上下文、资源通知知道它们的存在。 # 一个简单的做法是让execute_function在执行成功后告知LLM该工具已可用。 return [execute_schema, register_schema] app.call_tool() async def handle_call_tool(name: str, arguments: dict) - list: 处理工具调用请求。 if name execute_function: func_name arguments.get(name) func_args arguments.get(arguments, {}) if not func_name: return [{type: text, text: Error: name field is required for execute_function.}] # 委托给注册表执行 result_text registry.execute(func_name, func_args) return [{type: text, text: result_text}] elif name register_tool: # 实现动态注册逻辑 tool_name arguments.get(tool_name) description arguments.get(description) params_schema arguments.get(parameters_schema) source_type arguments.get(source_type) source_content arguments.get(source_content) # 安全性检查非常重要 if not all([tool_name, description, params_schema, source_type, source_content]): return [{type: text, text: Error: Missing required fields for register_tool.}] # 这里是一个简化的、极不安全的示例。生产环境必须使用沙箱 if source_type inline_python: try: # 警告直接exec是极度危险的仅用于演示概念。 # 生产环境应使用RestrictedPython、PySandbox等沙箱技术。 namespace {} exec(source_content, namespace) # 假设代码定义了一个名为tool_function的函数 func namespace.get(tool_function) if not callable(func): return [{type: text, text: Error: Inline code must define a callable named tool_function.}] # 注册到中心 registry.register(tool_name, func, params_schema) return [{type: text, text: fTool {tool_name} registered successfully from inline code.}] except Exception as e: return [{type: text, text: fError loading inline code: {e}}] elif source_type file_path: # 从文件加载模块的逻辑略同样需注意安全 return [{type: text, text: File path loading not implemented in this demo.}] else: return [{type: text, text: fError: Unsupported source_type {source_type}.}] else: # 理论上由于list_tools只返回两个核心工具不会走到这里。 # 但如果未来扩展了其他内置工具可以在这里处理。 return [{type: text, text: fError: Unknown tool {name}.}] async def main(): 运行MCP服务器 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, InitializationOptions( server_nameon-demand-mcp, server_version0.1.0, capabilitiesapp.get_capabilities( notification_optionsNotificationOptions(), experimental_capabilities{}, ), ), ) if __name__ __main__: asyncio.run(main())4.3 动态注册与调用演示现在我们的Server已经准备好了。启动它python server.py假设我们通过另一个进程或者预先在Server启动脚本里模拟Agent的初始操作动态注册一个get_weather工具。创建一个bootstrap.py脚本这模拟了Agent的初始化或按需加载行为# bootstrap.py import requests import json # 假设我们通过MCP Client协议与Server通信这里简化为直接调用注册逻辑 # 在实际中这可能是Server启动后立即执行的一段初始化代码或者由第一个LLM请求触发。 # 1. 定义我们要动态添加的工具get_weather weather_tool_code def tool_function(city: str) - str: \\\模拟获取城市天气信息。\\\ # 这里应该是真实的API调用例如调用和风天气、OpenWeatherMap等。 # 为演示我们返回模拟数据。 weather_data { Beijing: {temp: 22, condition: Sunny, humidity: 65}, Shanghai: {temp: 25, condition: Cloudy, humidity: 80}, Shenzhen: {temp: 28, condition: Rainy, humidity: 90}, } if city in weather_data: info weather_data[city] return fThe weather in {city} is {info[condition]} with a temperature of {info[temp]}°C and humidity {info[humidity]}%. else: return fWeather information for {city} is currently unavailable. # 2. 构建register_tool的调用参数 registration_request { tool_name: get_weather, description: 获取指定城市的当前天气信息。, parameters_schema: { type: object, properties: { city: { type: string, description: 城市名称例如 Beijing, Shanghai. } }, required: [city] }, source_type: inline_python, source_content: weather_tool_code } # 3. 在实际MCP通信中我们需要通过stdio发送一个tools/call请求来调用register_tool。 # 此处为演示我们直接导入registry并注册模拟Server内部初始化。 from tool_registry import registry import types # 动态执行代码获取函数对象 namespace {} exec(weather_tool_code, namespace) func namespace[tool_function] # 注册到全局注册表 registry.register( nameget_weather, funcfunc, schema{ name: get_weather, description: 获取指定城市的当前天气信息。, inputSchema: registration_request[parameters_schema] } ) print([Bootstrap] Tool get_weather has been dynamically registered.)在Server启动后运行这个bootstrap脚本需要确保Server进程中的Python解释器能导入到同一个tool_registry实例这在实际分布式部署中需要更精细的设计例如通过进程间通信或数据库。本例为单进程演示# 在另一个终端或者确保在Server启动前导入并运行 python bootstrap.py现在get_weather工具已经存在于ToolRegistry中但WebMCP Client如Claude Desktop通过list_tools仍然只能看到execute_function和register_tool。那么Agent该如何使用它呢这需要一点“约定”或“引导”。有两种常见策略策略A通过register_tool注册后立即通知。修改register_tool的实现使其在成功注册后不仅返回成功信息还“暗示”或“告知”LLM“工具X已就绪你现在可以通过execute_function工具传入nameX和相应参数来使用它。” 这可以通过在返回的文本中明确说明来实现。策略BAgent主动发现。我们可以提供第三个非MCP工具的发现机制。例如维护一个简单的文本文件或内存中的列表记录所有动态加载的工具名和描述。或者我们可以让execute_function在收到不认识的工具名时返回一个提示“工具未找到。当前已动态加载的工具有get_weather, ... 请先确认工具名或使用register_tool进行注册。”在实际的CreatorWeave项目中我采用了更智能的方式将ToolRegistry.list_tools()的结果也通过一个简单的MCPResource暴露出来。LLM可以首先读取这个资源来了解当前可用的动态工具列表然后再决定调用哪个。这样就形成了一个完整的闭环查询资源 - 按需注册 - 执行工具。假设我们添加了一个资源dynamic-tools://list其内容就是当前注册表中所有动态工具的JSON列表。那么LLM的工作流将是读取dynamic-tools://list资源发现没有get_weather。判断需要天气功能调用register_tool注册get_weather。再次读取dynamic-tools://list资源确认get_weather已存在。调用execute_functionname为get_weatherarguments为{city: Beijing}。5. 安全、性能与生产级考量上面的演示代码为了清晰省略了大量生产环境必需的细节。当你真正打算采用这种模式时以下几点至关重要。5.1 动态代码执行的安全性重中之重示例中直接使用exec()是极其危险的绝不能用于任何公开或生产环境。你必须建立严格的安全沙箱使用专用沙箱库考虑使用RestrictedPython、PySandbox注意其维护状态或docker容器来隔离执行环境。白名单机制只允许导入特定的、安全的模块如json,datetime,math禁止os,sys,subprocess,requests除非可控等。资源限制对执行时间、内存使用、磁盘IO进行严格限制。代码审计对要动态注册的代码进行简单的静态分析检查危险模式和关键字。签名与来源验证确保动态加载的代码来自可信源并且未被篡改。一个更安全的替代方案是不动态执行代码而是动态调用已部署的API。即register_tool注册的不是代码块而是一个API端点的描述URL、方法、参数映射。execute_function则变成一个安全的HTTP客户端去调用那个API。这样核心Server完全不执行未知代码所有业务逻辑都在受控的API服务中。5.2 工具发现与通信优化资源通知Resource Notifications这是MCP协议的原生支持。当新工具被注册时Server可以主动向Client发送一个notify消息告知有一个新的resource可以理解为工具目录可用。Client收到后可以去读取这个资源从而更新其工具列表。这比轮询更高效。工具Schema的缓存与同步LLMClient需要知道工具的详细参数schema才能正确调用。在动态注册后需要有一种机制将新的schema同步给Client。除了通过Resource也可以考虑在register_tool成功后通过MCP的notify机制推送一个“工具列表已更新”的事件引导Client重新调用list_tools此时需要修改list_tools以包含动态工具。5.3 状态管理与持久化在单进程演示中我们用了内存字典。在生产中工具注册表需要持久化使用数据库如SQLite、PostgreSQL或分布式缓存如Redis来存储工具定义以便Server重启后不丢失。会话隔离不同的用户或会话可能拥有不同的动态工具集。需要在注册和查询时加入会话ID或用户ID作为命名空间。版本管理同一个工具可能有多个版本需要妥善处理。5.4 错误处理与用户体验友好的错误信息execute_function的返回信息需要足够友好不仅能给LLM解析最好也能让最终用户理解。例如工具执行失败时应返回结构化的错误信息包括错误类型、建议的解决步骤等。工具依赖与初始化有些工具可能需要初始化如加载大模型、连接数据库。在动态注册时需要考虑如何管理这些依赖和初始化过程。超时与重试对于代理外部API的工具必须有超时和重试机制防止一个缓慢的请求阻塞整个Agent。6. 在CreatorWeave中的真实应用与扩展在CreatorWeave一个AI辅助内容创作平台中我将On-Demand模式运用到了几个具体场景场景一插件化内容处理管道。我们有两个核心工具apply_content_plugin应用内容插件和list_available_plugins列出可用插件。每个“插件”就是一个动态工具比如“标题优化插件”、“SEO关键词提取插件”、“多语言翻译插件”。当用户要求“优化这篇文章的标题”时Agent会先检查插件列表如果没有“标题优化”插件则从插件市场动态加载其定义并注册然后调用apply_content_plugin执行它。这使得平台的功能可以无限扩展而核心Agent架构保持不变。场景二外部数据源连接器。用户经常需要接入自己的数据源如Notion数据库、Google Sheets、企业内部CMS。我们提供了一个connect_external_source连接外部源工具。用户通过自然语言描述数据源类型和认证信息Agent调用此工具该工具会在后端动态生成一个针对该数据源的专用查询工具例如query_notion_db_XXXX并注册到当前会话中。之后用户就可以直接说“从刚才连接的Notion里找出上周的会议纪要”。扩展思路工具组合与工作流生成。这是On-Demand模式的终极形态。Agent不仅能够按需加载工具还能根据复杂任务的目标自动将多个已注册或可注册的工具组合成一个临时的工作流即一个更高阶的“工具”并动态注册它。例如用户说“帮我分析这个产品反馈文档总结痛点并生成一个功能脑图”。Agent可以规划出步骤1. 调用文本摘要工具。2. 调用情感分析工具。3. 调用脑图生成工具。然后它可以将这三个步骤封装成一个新的复合工具analyze_feedback_and_generate_mindmap并动态注册最后执行它。这实现了真正的“任务驱动”的自动化。回过头看“只注册2个工具代理N个”不仅仅是一个节省配置的技巧它代表了一种构建AI Agent系统的范式转变从静态、封闭的工具箱转向动态、开放的能力网络。它要求我们将设计重心从“提供所有功能”转移到“提供发现、集成和执行功能的核心机制”上。这种架构为AI Agent带来了前所未有的适应性和扩展性让它更像一个真正的“智能体”能够学习新技能而不仅仅是被动地使用预设功能。当然这种强大能力也伴随着安全性和复杂性的挑战需要在设计之初就慎重考虑。希望这篇来自CreatorWeave实践中的分享能为你设计自己的WebMCP应用打开一扇新的大门。 WebMCP On-Demand工具注册：用2个核心工具代理N个功能的架构实践
1. 从“工具海”到“工具池”一个被忽视的WebMCP核心范式如果你最近在折腾AI Agent或者LLM应用开发大概率听说过WebMCPModel Context Protocol这个协议。它被设计为连接LLM与外部工具和数据的桥梁听起来很美好但上手后很多人会陷入一个困境为了一个功能丰富的Agent我是不是得预先注册几十上百个工具配置文件越来越臃肿启动越来越慢维护成本直线上升这和我们追求的灵活、智能的Agent似乎背道而驰。这正是我最初接触WebMCP时的困惑。直到我在CreatorWeave这个项目中实践了“On-Demand”按需工具注册模式才真正理解了WebMCP设计哲学中一个被严重低估的侧面。它的精髓不在于你能注册多少工具而在于你能否用最少的、最核心的工具去动态地、智能地“代理”出无限可能的功能。这篇文章我就来拆解这个“只注册2个工具代理N个”的实践它不仅仅是技术实现更是一种架构思维的转变。简单来说我们不再把Agent看作一个拥有固定技能列表的“瑞士军刀”而是把它看作一个拥有核心“元能力”的“指挥官”。这个指挥官手里只有两把万能钥匙两个核心注册工具但它知道如何根据任务指令动态地调用、组合外部资源即被代理的N个工具或服务来完成任务。这极大地提升了Agent的灵活性、可维护性和响应速度。无论你是想构建一个能处理复杂工作流的创作助手还是一个能连接企业内部各种API的业务Agent这个思路都能帮你跳出“工具注册地狱”。2. 为什么是“On-Demand”重新审视WebMCP的工具管理在深入代码之前我们必须先厘清一个根本问题为什么传统的“预注册所有工具”模式在复杂场景下会失效以及WebMCP协议本身是否支持更动态的方式2.1 传统预注册模式的三大痛点第一启动与内存开销。一个WebMCP Server在启动时需要加载并初始化所有注册的工具函数、它们的schemaJSON Schema描述。如果工具数量庞大例如超过50个这个初始化过程会显著拖慢启动速度并且这些工具的定义会常驻内存即使整个会话周期一次都用不到。第二配置与维护的复杂性。mcp_server的配置文件如server.py或config.json会变得极其冗长。每增加、删除或修改一个工具都需要更新这个配置文件并重启服务。在快速迭代的开发阶段或者工具集动态变化的场景下这简直是噩梦。第三功能暴露的“噪音”。当Agent如Claude Desktop、Cursor等连接到你的Server时它会收到一个包含所有工具签名的庞大列表。这可能会干扰LLM的决策让它陷入“选择困难症”或者在不合适的时机调用不相关的工具。2.2 WebMCP协议提供的灵活性工具Tools与资源ResourcesWebMCP协议的设计其实已经考虑到了动态性。它不仅仅有Tools工具还有Resources资源。Tools代表可执行的操作函数而Resources代表可读取的数据或状态如文件内容、数据库查询结果。协议允许Server在运行时动态地公布notify新的Resources给Client。虽然协议标准更侧重于Resources的动态性但Tools的动态注册与注销在实现层面并非不可能。许多MCP Server的实现包括官方和社区的都提供了运行时更新工具列表的接口。“On-Demand”模式的核心思想就是利用这种潜力将工具的“注册”动作延迟到真正需要它的前一刻或者通过一个“代理工具”来间接调用。2.3 “On-Demand”模式的核心优势对比之下On-Demand模式的优势就非常明显了极简启动Server启动时只加载最核心、最通用的几个工具例如2个启动飞快内存占用小。动态能力Agent的能力边界不再是固定的可以根据用户请求的上下文动态地加载或指向新的功能模块。解耦与维护新增功能模块即被代理的N个工具可以独立开发、测试、部署只需确保它们能被核心的“代理工具”访问到即可无需修改主Server的配置。清晰的架构两个核心工具充当了明确的“网关”或“路由器”使得整个系统的数据流和控制流更加清晰。3. 核心设计哪两个工具足以“代理万物”这是整个模式最巧妙也最需要深思熟虑的部分。选择哪两个工具作为“万能钥匙”决定了整个架构的形态和易用性。在CreatorWeave的实践中我经过多次迭代最终确定了两个方向一个用于“执行”一个用于“发现与管理”。这里我提供两种经过验证的设计方案。3.1 方案一函数执行器 工具注册器经典网关模式这是最直接、最强大的模式类似于一个微型的“云函数”平台或“插件系统”。execute_function函数执行器职责接收一个函数名和参数在安全沙箱或特定上下文中执行对应的函数并返回结果。工作原理Server背后维护着一个工具函数注册表例如一个Python字典或数据库。这个注册表最初是空的或者只包含少数内置函数。execute_function工具被调用时它根据传入的function_name去注册表中查找对应的函数对象然后用arguments参数调用它。如何实现“代理N个”那“N个”工具的实现就是一个个独立的函数它们被预先编写好存储在一个特定的目录如./tools/或模块中。当需要某个工具时可以通过另一个工具或初始化脚本将这些函数动态地“注入”到注册表中。execute_function本身并不关心它执行的是哪个函数它只是一个安全的调用网关。# 伪代码示例execute_function 的核心逻辑 class ToolRegistry: _functions {} # 全局函数注册表 classmethod def register(cls, name: str, func: callable, schema: dict): cls._functions[name] {func: func, schema: schema} classmethod def execute(cls, name: str, arguments: dict) - str: if name not in cls._functions: return fError: Function {name} not found. tool_info cls._functions[name] try: result tool_info[func](**arguments) return str(result) except Exception as e: return fError executing {name}: {e} # WebMCP Tool 定义 async def execute_function(name: str, arguments: dict) - str: return ToolRegistry.execute(name, arguments)register_tool工具注册器职责接收一个新工具的定义包括名称、描述、参数schema和函数引用将其动态添加到ToolRegistry中。工作原理这是实现“On-Demand”的关键。Agent在运行过程中如果判断需要某个尚未加载的工具它可以先调用register_tool将工具的定义从磁盘、网络或代码字符串中加载并注册然后再调用execute_function来使用它。更高级的玩法是Agent可以分析用户需求自动生成工具的描述和参数schema然后注册一个“虚拟工具”其执行函数可能是一个对LLM的二次调用或一个复杂的工作流。# 伪代码示例register_tool 工具 async def register_tool(tool_name: str, description: str, parameters_schema: dict, source_code_or_path: str) - str: # 1. 验证schema # 2. 从source_code_or_path加载或编译函数需要安全考虑 # 3. 将函数对象和schema注册到ToolRegistry ToolRegistry.register(tool_name, loaded_function, parameters_schema) return fTool {tool_name} registered successfully.这个方案的强大之处在于它将工具的“声明”和“执行”完全分离并且将“注册”本身也工具化了。Agent获得了在运行时扩展自身能力的可能性。3.2 方案二工作流执行器 资源查询器面向流程模式如果你的N个工具通常是按照特定顺序、为了完成一个特定目标如“写博客”、“分析数据”而组合使用的那么这种模式更合适。它更侧重于编排而非单个工具的动态注册。run_workflow工作流执行器职责接收一个工作流标识符如workflow_id和输入参数执行一个预定义的工作流。工作原理工作流是一个有向无环图DAG节点是一个个具体的工具或操作即那“N个”工具边定义了执行顺序和参数传递。run_workflow工具背后连接着一个工作流引擎可以是简单的Python脚本也可以是像Prefect、Airflow这样的框架。当被调用时它启动对应的工作流并管理其中各个节点的执行。这里的“N个工具”是工作流内部的固定步骤但对WebMCP Server来说它只暴露了run_workflow这一个执行接口。query_available_workflows资源查询器职责列出当前所有可用的工作流及其描述、输入参数。工作原理它扫描工作流定义目录或者查询工作流引擎的元数据然后以结构化数据如JSON的形式返回。这相当于一个动态的“能力目录”帮助LLM了解当前可以发起哪些复杂的任务。这个方案的优势是对于复杂、多步骤的任务封装性好用户体验流畅。用户或Agent只需要说“帮我执行数据分析流程”而不需要关心内部调用了多少个API、做了多少次数据转换。劣势是工作流需要预先定义动态创建和修改工作流本身比较复杂。在CreatorWeave中我主要采用了方案一函数执行器注册器因为它提供了最大的灵活性允许Agent在对话中实时“学习”并使用新工具。接下来我们就看看如何具体实现它。4. 实战构建一个极简On-Demand WebMCP Server让我们用Python和mcpSDK快速搭建一个原型。假设我们的目标是构建一个Server它启动时只注册execute_function和register_tool。然后我们通过一个外部命令或Agent的初始请求动态加载一个get_weather获取天气工具并成功调用它。4.1 项目初始化与基础结构首先创建项目并安装依赖。mkdir on-demand-mcp-server cd on-demand-mcp-server python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install mcp创建主服务器文件server.py和工具注册表模块tool_registry.py。tool_registry.py内容如下# tool_registry.py import importlib.util import sys import json from typing import Dict, Any, Callable, Optional class ToolRegistry: 全局工具注册中心采用单例模式简化 _instance None _tools: Dict[str, Dict[str, Any]] {} # {tool_name: {func: callable, schema: dict}} def __new__(cls): if cls._instance is None: cls._instance super(ToolRegistry, cls).__new__(cls) return cls._instance def register(self, name: str, func: Callable, schema: dict): 注册一个工具到中心仓库 if name in self._tools: print(f[Warning] Tool {name} is being overwritten.) self._tools[name] {func: func, schema: schema} print(f[Info] Tool {name} registered.) def get(self, name: str) - Optional[Dict[str, Any]]: 根据名称获取工具信息 return self._tools.get(name) def list_tools(self) - Dict[str, dict]: 列出所有已注册的工具返回schema供MCP使用 return {name: info[schema] for name, info in self._tools.items()} def execute(self, name: str, arguments: dict) - str: 执行指定工具 tool_info self.get(name) if not tool_info: return json.dumps({error: fTool {name} not found in registry.}) try: # 调用实际的函数 result tool_info[func](**arguments) # 将结果转换为字符串对于复杂对象可以JSON序列化 if isinstance(result, (dict, list)): return json.dumps(result, ensure_asciiFalse) return str(result) except TypeError as e: return json.dumps({error: fInvalid arguments for {name}: {e}}) except Exception as e: return json.dumps({error: fExecution error for {name}: {e}}) # 全局注册表实例方便导入 registry ToolRegistry()4.2 实现核心的“两个工具”现在在server.py中我们基于tool_registry实现那两个核心的MCP工具。# server.py import asyncio from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import json from tool_registry import registry # 创建MCP Server实例 app Server(on-demand-mcp-server) app.list_tools() async def handle_list_tools(): 列出当前已注册的所有工具。这里只返回两个核心工具。 # 核心工具1: execute_function execute_schema { name: execute_function, description: 执行一个在注册表中已注册的工具函数。, inputSchema: { type: object, properties: { name: { type: string, description: 要执行的工具函数名称。 }, arguments: { type: object, description: 传递给工具函数的参数字典。, additionalProperties: True # 参数取决于具体工具 } }, required: [name, arguments] } } # 核心工具2: register_tool register_schema { name: register_tool, description: 动态注册一个新的工具到服务器。工具定义需包含可执行的代码或路径。, inputSchema: { type: object, properties: { tool_name: { type: string, description: 新工具的唯一名称。 }, description: { type: string, description: 工具的功能描述。 }, parameters_schema: { type: object, description: 符合JSON Schema的工具参数定义。, additionalProperties: True }, source_type: { type: string, enum: [inline_python, file_path], description: 工具源码的提供方式。 }, source_content: { type: string, description: 根据source_type可以是Python代码字符串或文件路径。 } }, required: [tool_name, description, parameters_schema, source_type, source_content] } } # 注意这里我们只返回这两个核心工具。 # 被动态注册的工具如get_weather不会在这里直接列出。 # 但LLM可以通过其他方式如对话上下文、资源通知知道它们的存在。 # 一个简单的做法是让execute_function在执行成功后告知LLM该工具已可用。 return [execute_schema, register_schema] app.call_tool() async def handle_call_tool(name: str, arguments: dict) - list: 处理工具调用请求。 if name execute_function: func_name arguments.get(name) func_args arguments.get(arguments, {}) if not func_name: return [{type: text, text: Error: name field is required for execute_function.}] # 委托给注册表执行 result_text registry.execute(func_name, func_args) return [{type: text, text: result_text}] elif name register_tool: # 实现动态注册逻辑 tool_name arguments.get(tool_name) description arguments.get(description) params_schema arguments.get(parameters_schema) source_type arguments.get(source_type) source_content arguments.get(source_content) # 安全性检查非常重要 if not all([tool_name, description, params_schema, source_type, source_content]): return [{type: text, text: Error: Missing required fields for register_tool.}] # 这里是一个简化的、极不安全的示例。生产环境必须使用沙箱 if source_type inline_python: try: # 警告直接exec是极度危险的仅用于演示概念。 # 生产环境应使用RestrictedPython、PySandbox等沙箱技术。 namespace {} exec(source_content, namespace) # 假设代码定义了一个名为tool_function的函数 func namespace.get(tool_function) if not callable(func): return [{type: text, text: Error: Inline code must define a callable named tool_function.}] # 注册到中心 registry.register(tool_name, func, params_schema) return [{type: text, text: fTool {tool_name} registered successfully from inline code.}] except Exception as e: return [{type: text, text: fError loading inline code: {e}}] elif source_type file_path: # 从文件加载模块的逻辑略同样需注意安全 return [{type: text, text: File path loading not implemented in this demo.}] else: return [{type: text, text: fError: Unsupported source_type {source_type}.}] else: # 理论上由于list_tools只返回两个核心工具不会走到这里。 # 但如果未来扩展了其他内置工具可以在这里处理。 return [{type: text, text: fError: Unknown tool {name}.}] async def main(): 运行MCP服务器 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, InitializationOptions( server_nameon-demand-mcp, server_version0.1.0, capabilitiesapp.get_capabilities( notification_optionsNotificationOptions(), experimental_capabilities{}, ), ), ) if __name__ __main__: asyncio.run(main())4.3 动态注册与调用演示现在我们的Server已经准备好了。启动它python server.py假设我们通过另一个进程或者预先在Server启动脚本里模拟Agent的初始操作动态注册一个get_weather工具。创建一个bootstrap.py脚本这模拟了Agent的初始化或按需加载行为# bootstrap.py import requests import json # 假设我们通过MCP Client协议与Server通信这里简化为直接调用注册逻辑 # 在实际中这可能是Server启动后立即执行的一段初始化代码或者由第一个LLM请求触发。 # 1. 定义我们要动态添加的工具get_weather weather_tool_code def tool_function(city: str) - str: \\\模拟获取城市天气信息。\\\ # 这里应该是真实的API调用例如调用和风天气、OpenWeatherMap等。 # 为演示我们返回模拟数据。 weather_data { Beijing: {temp: 22, condition: Sunny, humidity: 65}, Shanghai: {temp: 25, condition: Cloudy, humidity: 80}, Shenzhen: {temp: 28, condition: Rainy, humidity: 90}, } if city in weather_data: info weather_data[city] return fThe weather in {city} is {info[condition]} with a temperature of {info[temp]}°C and humidity {info[humidity]}%. else: return fWeather information for {city} is currently unavailable. # 2. 构建register_tool的调用参数 registration_request { tool_name: get_weather, description: 获取指定城市的当前天气信息。, parameters_schema: { type: object, properties: { city: { type: string, description: 城市名称例如 Beijing, Shanghai. } }, required: [city] }, source_type: inline_python, source_content: weather_tool_code } # 3. 在实际MCP通信中我们需要通过stdio发送一个tools/call请求来调用register_tool。 # 此处为演示我们直接导入registry并注册模拟Server内部初始化。 from tool_registry import registry import types # 动态执行代码获取函数对象 namespace {} exec(weather_tool_code, namespace) func namespace[tool_function] # 注册到全局注册表 registry.register( nameget_weather, funcfunc, schema{ name: get_weather, description: 获取指定城市的当前天气信息。, inputSchema: registration_request[parameters_schema] } ) print([Bootstrap] Tool get_weather has been dynamically registered.)在Server启动后运行这个bootstrap脚本需要确保Server进程中的Python解释器能导入到同一个tool_registry实例这在实际分布式部署中需要更精细的设计例如通过进程间通信或数据库。本例为单进程演示# 在另一个终端或者确保在Server启动前导入并运行 python bootstrap.py现在get_weather工具已经存在于ToolRegistry中但WebMCP Client如Claude Desktop通过list_tools仍然只能看到execute_function和register_tool。那么Agent该如何使用它呢这需要一点“约定”或“引导”。有两种常见策略策略A通过register_tool注册后立即通知。修改register_tool的实现使其在成功注册后不仅返回成功信息还“暗示”或“告知”LLM“工具X已就绪你现在可以通过execute_function工具传入nameX和相应参数来使用它。” 这可以通过在返回的文本中明确说明来实现。策略BAgent主动发现。我们可以提供第三个非MCP工具的发现机制。例如维护一个简单的文本文件或内存中的列表记录所有动态加载的工具名和描述。或者我们可以让execute_function在收到不认识的工具名时返回一个提示“工具未找到。当前已动态加载的工具有get_weather, ... 请先确认工具名或使用register_tool进行注册。”在实际的CreatorWeave项目中我采用了更智能的方式将ToolRegistry.list_tools()的结果也通过一个简单的MCPResource暴露出来。LLM可以首先读取这个资源来了解当前可用的动态工具列表然后再决定调用哪个。这样就形成了一个完整的闭环查询资源 - 按需注册 - 执行工具。假设我们添加了一个资源dynamic-tools://list其内容就是当前注册表中所有动态工具的JSON列表。那么LLM的工作流将是读取dynamic-tools://list资源发现没有get_weather。判断需要天气功能调用register_tool注册get_weather。再次读取dynamic-tools://list资源确认get_weather已存在。调用execute_functionname为get_weatherarguments为{city: Beijing}。5. 安全、性能与生产级考量上面的演示代码为了清晰省略了大量生产环境必需的细节。当你真正打算采用这种模式时以下几点至关重要。5.1 动态代码执行的安全性重中之重示例中直接使用exec()是极其危险的绝不能用于任何公开或生产环境。你必须建立严格的安全沙箱使用专用沙箱库考虑使用RestrictedPython、PySandbox注意其维护状态或docker容器来隔离执行环境。白名单机制只允许导入特定的、安全的模块如json,datetime,math禁止os,sys,subprocess,requests除非可控等。资源限制对执行时间、内存使用、磁盘IO进行严格限制。代码审计对要动态注册的代码进行简单的静态分析检查危险模式和关键字。签名与来源验证确保动态加载的代码来自可信源并且未被篡改。一个更安全的替代方案是不动态执行代码而是动态调用已部署的API。即register_tool注册的不是代码块而是一个API端点的描述URL、方法、参数映射。execute_function则变成一个安全的HTTP客户端去调用那个API。这样核心Server完全不执行未知代码所有业务逻辑都在受控的API服务中。5.2 工具发现与通信优化资源通知Resource Notifications这是MCP协议的原生支持。当新工具被注册时Server可以主动向Client发送一个notify消息告知有一个新的resource可以理解为工具目录可用。Client收到后可以去读取这个资源从而更新其工具列表。这比轮询更高效。工具Schema的缓存与同步LLMClient需要知道工具的详细参数schema才能正确调用。在动态注册后需要有一种机制将新的schema同步给Client。除了通过Resource也可以考虑在register_tool成功后通过MCP的notify机制推送一个“工具列表已更新”的事件引导Client重新调用list_tools此时需要修改list_tools以包含动态工具。5.3 状态管理与持久化在单进程演示中我们用了内存字典。在生产中工具注册表需要持久化使用数据库如SQLite、PostgreSQL或分布式缓存如Redis来存储工具定义以便Server重启后不丢失。会话隔离不同的用户或会话可能拥有不同的动态工具集。需要在注册和查询时加入会话ID或用户ID作为命名空间。版本管理同一个工具可能有多个版本需要妥善处理。5.4 错误处理与用户体验友好的错误信息execute_function的返回信息需要足够友好不仅能给LLM解析最好也能让最终用户理解。例如工具执行失败时应返回结构化的错误信息包括错误类型、建议的解决步骤等。工具依赖与初始化有些工具可能需要初始化如加载大模型、连接数据库。在动态注册时需要考虑如何管理这些依赖和初始化过程。超时与重试对于代理外部API的工具必须有超时和重试机制防止一个缓慢的请求阻塞整个Agent。6. 在CreatorWeave中的真实应用与扩展在CreatorWeave一个AI辅助内容创作平台中我将On-Demand模式运用到了几个具体场景场景一插件化内容处理管道。我们有两个核心工具apply_content_plugin应用内容插件和list_available_plugins列出可用插件。每个“插件”就是一个动态工具比如“标题优化插件”、“SEO关键词提取插件”、“多语言翻译插件”。当用户要求“优化这篇文章的标题”时Agent会先检查插件列表如果没有“标题优化”插件则从插件市场动态加载其定义并注册然后调用apply_content_plugin执行它。这使得平台的功能可以无限扩展而核心Agent架构保持不变。场景二外部数据源连接器。用户经常需要接入自己的数据源如Notion数据库、Google Sheets、企业内部CMS。我们提供了一个connect_external_source连接外部源工具。用户通过自然语言描述数据源类型和认证信息Agent调用此工具该工具会在后端动态生成一个针对该数据源的专用查询工具例如query_notion_db_XXXX并注册到当前会话中。之后用户就可以直接说“从刚才连接的Notion里找出上周的会议纪要”。扩展思路工具组合与工作流生成。这是On-Demand模式的终极形态。Agent不仅能够按需加载工具还能根据复杂任务的目标自动将多个已注册或可注册的工具组合成一个临时的工作流即一个更高阶的“工具”并动态注册它。例如用户说“帮我分析这个产品反馈文档总结痛点并生成一个功能脑图”。Agent可以规划出步骤1. 调用文本摘要工具。2. 调用情感分析工具。3. 调用脑图生成工具。然后它可以将这三个步骤封装成一个新的复合工具analyze_feedback_and_generate_mindmap并动态注册最后执行它。这实现了真正的“任务驱动”的自动化。回过头看“只注册2个工具代理N个”不仅仅是一个节省配置的技巧它代表了一种构建AI Agent系统的范式转变从静态、封闭的工具箱转向动态、开放的能力网络。它要求我们将设计重心从“提供所有功能”转移到“提供发现、集成和执行功能的核心机制”上。这种架构为AI Agent带来了前所未有的适应性和扩展性让它更像一个真正的“智能体”能够学习新技能而不仅仅是被动地使用预设功能。当然这种强大能力也伴随着安全性和复杂性的挑战需要在设计之初就慎重考虑。希望这篇来自CreatorWeave实践中的分享能为你设计自己的WebMCP应用打开一扇新的大门。