swarms 框架 MCPDeployer 实战指南将 Agent 与任意 Swarm 一键部署为带鉴权的 MCP 服务器【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms导读MCPDeployer是 swarms 框架中负责服务化的核心组件它能把一个Agent、任何具备run()方法的 swarm例如SequentialWorkflow、SwarmRouter甚至一个普通 Python 函数包装成符合 MCPModel Context Protocol规范的服务器并在每个 HTTP 请求进入传输层之前插入一层鉴权。本文以 examples/mcp/mcp_deployer/README.md 为主线结合 swarms/structs/mcp_deployer.py 源码与 11 个示例文件系统讲解部署方式、鉴权优先级、生命周期管理、三种传输协议以及如何在另一个 Agent 中通过MCPConnection/MCPManager消费这些服务。读完你可以在几分钟内把任意 Agent 变成带 API Key、租户隔离或 OAuth 风格 Token 校验的远程 MCP 工具。一、MCPDeployer 是什么一个目标 一个 MCP 工具MCPDeployer的核心模型非常简单每一个被部署的目标target都会成为服务器上的一个 MCP 工具tool工具名取自目标名输入 schema 统一为(task, img)——task是任务字符串img是可选图片参数默认为None。可被部署的目标需要满足_is_servable的判定见 swarms/structs/mcp_deployer.py 第 96-98 行目标要么可调用callable要么具备可调用的run方法。因此以下对象都可以直接部署单个Agent任何带run(task, ...)方法的 swarm 结构SequentialWorkflow、SwarmRouter、HeavySwarm等接收任务字符串的普通 Python 函数。1.1 最简部署一个 AgentREADME 给出的最小可用代码只有三行from swarms import Agent, MCPDeployer agent Agent(agent_nameResearcher, model_namegpt-5.4, max_loops1) MCPDeployer(agent, api_keys[sk-local-dev], port8000).run()运行后服务器会在127.0.0.1:8000上监听MCP 端点默认为/mcp。另一个 Agent 只需这样连接Agent(mcp_urlMCPConnection(urlhttp://127.0.0.1:8000/mcp, api_keysk-local-dev))1.2 工具名的自动推导如果不显式指定工具名MCPDeployer会按照_default_tool_nameswarms/structs/mcp_deployer.py 第 100-109 行自动生成一个 snake_case 名称取值优先级为agent_name→name→__name__→ 类型名。例如名为Researcher的 Agent 会暴露为工具researcher。工具描述description同样自动推导_default_description第 112-120 行优先取agent_description或description属性否则取 docstring 首行最后兜底为Run tool_name on a task.。1.3 单服务器承载多个目标把目标传成list或dict即可在一个服务器上同时暴露多个工具。dict 的 key 就是工具名见 swarms/structs/mcp_deployer.py 第 342-360 行MCPDeployer( {research: researcher, write: writer, review: pipeline}, api_keys[sk-local-dev], ).run()注意传入 dict 或 list 时不能再传tool_name或description那仅适用于单目标场景否则构造函数会抛出ValueError。服务器启动后还可以通过add_tool(target, name, description)在run()/start()之前动态注册更多工具。1.4 部署目标的内部调用链从源码_invokeswarms/structs/mcp_deployer.py 第 525-538 行可以看出每次工具调用的执行逻辑若目标有run方法且客户端传了img先尝试run(task, imgimg)遇到TypeError则回退有run方法则调用run(task)否则直接把目标当函数调用target(task)。返回值统一经过_to_text第 123-132 行渲染为字符串字符串原样返回其他对象dict、list 等用json.dumps(result, indent2, defaultstr)序列化。也就是说函数目标返回任意可序列化对象都可以客户端拿到的永远是结构化文本结果。二、示例全景11 个可复制的实战脚本examples/mcp/mcp_deployer 目录下共 11 个示例覆盖了不同的目标类型、鉴权方式与传输协议下面的总表来自 README 并已核对各脚本实际内容示例目标鉴权传输协议是否独立运行single_agent_api_key.py单个Agent静态api_keysstreamable HTTP持续服务直到被中断sequential_workflow_as_tool.pySequentialWorkflow静态api_keysstreamable HTTP持续服务直到被中断multiple_agents_one_server.py两个 Agent、一个SequentialWorkflow、两个函数各自独立成工具静态api_keysstreamable HTTP持续服务直到被中断custom_auth_per_tenant.py单个Agent异步auth回调读取x-tenantstreamable HTTP持续服务直到被中断owner_key_or_tenant_auth.py单个Agent同步auth回调环境变量中的 owner key或白名单x-tenantstreamable HTTP持续服务直到被中断token_verifier_with_scopes.py单个AgentTokenVerifier协议 required_scopesstreamable HTTP持续服务直到被中断env_keys_and_extra_tools.py单个Agent外加两个普通函数api_key_envstreamable HTTP持续服务直到被中断background_server_and_client_agent.py单个Agent静态api_keysstreamable HTTP是后台服务、调用、自动停止plain_function_json_response.py纯函数无 LLM静态api_keysstreamable HTTPJSON 响应是无需任何 LLM Keysse_transport.py单个Agent静态api_keysSSE持续服务直到被中断stdio_transport.py单个Agent无宿主即边界stdio由 MCP 宿主host拉起所有示例的模型名都使用普通的 LiteLLM 字符串如gpt-5.4、claude-sonnet-4-6、openrouter/moonshotai/kimi-k3你可以随意替换成任何你拥有 API Key 的厂商模型不需要改动其他任何代码。三、鉴权体系优先级明确的四层防护MCPDeployer最与众不同的能力是把鉴权层放在 MCP 传输层之前。从源码实现看_AuthMiddlewareswarms/structs/mcp_deployer.py 第 166-203 行是一个 ASGI 中间件它对每个 HTTP 请求先调用deployer.authenticate(headers)只有返回AuthResult的请求才会被放行到内部的 MCP 应用被拒绝的请求返回401并附带WWW-Authenticate: Bearer响应头。鉴权检查的顺序authenticate方法第 459-519 行与 README 的Auth, in order of precedence完全一致3.1 第一优先级authcallable(credential, headers)你自己的校验函数同步或异步均可签名是(credential, headers) - bool | dict | None返回真值truthy→ 放行返回dict→ 放行并把这个 dict 作为该请求的 claims声明可供后续工具调用上下文使用返回假值或抛出异常 → 拒绝。从源码第 481-493 行可以看到auth返回的 dict 会被包装成AuthResult其中subject取claims.get(subject) or claims.get(sub)scopes取claims.get(scopes, [])。owner_key_or_tenant_auth.py 展示了最典型的多重判断凭据等于环境变量MCP_DEPLOYER_KEY时以owner身份放行或者x-tenant头在白名单{acme, globex}内时按租户放行否则拒绝def is_allowed(credential, headers): if credential and credential os.getenv(MCP_DEPLOYER_KEY): return {subject: owner, scopes: [run]} tenant headers.get(x-tenant) if tenant in {acme, globex}: return {subject: tenant, scopes: [run]} return False deployer MCPDeployer(researcher, port8000, authis_allowed, timeout120, verboseTrue)custom_auth_per_tenant.py 则展示了异步版本每个租户持有独立 Key凭据与TENANT_KEYS[tenant]匹配后才放行并把{subject: tenant, scopes: [run], tenant: tenant}作为 claims 写入请求。这就是标准的一租户一密钥多租户隔离方案。3.2 第二优先级token_verifierOAuth 风格 Token当未配置auth时可以传入mcp包定义的TokenVerifier协议实现。authenticate会调用await token_verifier.verify_token(credential)并强制校验两点源码第 498-515 行过期时间token.expires_at若小于当前时间则拒绝作用域required_scopes中的每一项都必须出现在 token 的scopes里缺一即拒绝。token_verifier_with_scopes.py 给出了完整示例签发两个 tokentok-admin带[agent:run, agent:admin]且一小时内有效tok-reader只有[agent:read]注释明确说明它会因缺少agent:run而被 401 拒绝class StaticTokenVerifier: async def verify_token(self, token: str): return ISSUED_TOKENS.get(token) deployer MCPDeployer( summariser, token_verifierStaticTokenVerifier(), required_scopes[agent:run], port8003, )3.3 第三优先级api_keys[...]与api_key_envVAR静态 Keyapi_keys直接传入的静态 Key 列表api_key_env环境变量名变量值为逗号分隔的多个 Key在构造时读取_collect_api_keys第 441-451 行因此可以写成MCP_DEPLOYER_KEYSkey-a,key-b。静态 Key 使用hmac.compare_digest进行常量时间比较第 453-457 行避免时序侧信道攻击。需要留意环境变量读取发生在MCPDeployer(...)构造那一刻运行时再修改环境变量不会生效。3.4 第四优先级allow_anonymousTrue显式匿名匿名访问必须显式开启。源码第 314-323 行有一条强约束如果allow_anonymous为假且auth、token_verifier、api_keys全部未配置构造函数直接抛出ValueError杜绝裸奔上线。3.5 客户端如何携带凭据README 明确说明两种方式专用头x-api-key可用api_key_header参数改名标准Authorization: Bearer token这正是MCPManager默认发送的方式。credential_from_headers源码第 135-163 行的解析逻辑是先读api_key_header默认x-api-key带Bearer前缀时自动剥掉读不到再退回Authorization: Bearer。客户端侧的MCPConnection定义于 swarms/schemas/mcp_schemas.py 第 111-159 行默认api_key_headerAuthorization、api_key_prefixBearer与服务器端默认完全对齐。3.6 公共路径与健康检查public_paths默认为(/health,)第 65 行即/health永远公开。_build_server里注册的健康检查路由第 566-576 行会返回status、服务器name、当前tool、全部tools列表与transport类型非常适合接入负载均衡器或 K8s 探针。四、生命周期管理阻塞、后台线程与上下文管理器MCPDeployer提供三种运行方式README 的 Lifecycle 一节run()阻塞式持续服务直到 CtrlC 中断内部走uvicorn.run见源码第 726-746 行start()/stop()把 uvicorn 服务器放进后台线程运行start()会等待 socket 打开后返回wait参数控制最长等待秒数默认 10 秒超时会抛RuntimeError见第 748-787 行with MCPDeployer(...) as d:上下文管理器进入时start()、退出时自动stop()第 798-802 行。background_server_and_client_agent.py 把后台服务 客户端调用 自动停止完整串了起来with MCPDeployer(poet, api_keys[demo-key], portfree_port()) as server: print(fserving {server.tool_name} at {server.url}) client Agent( agent_nameClient, system_promptUse the poet tool to get a poem, then return it verbatim., model_nameclaude-haiku-4-5-20251001, mcp_urlMCPConnection(urlserver.url, api_keydemo-key), max_loops2, print_onFalse, output_typefinal, ) client.run(Get a poem about lighthouses.) print(poet.short_memory.get_final_message_content()) # 离开 with 块即停止服务器这里还展示了两个实用细节server.url属性会拼出完整的端点地址http://host:port/path源码第 596-598 行服务端 Agent 的结果可以通过short_memory.get_final_message_content()从内存中取回。4.1 单次工具调用的超时控制timeout参数以秒为单位约束单次工具调用的最大执行时间。源码第 547-551 行用anyio.fail_after(self.timeout)包裹同步调用通过anyio.to_thread.run_sync丢到线程池执行。对串联多次模型调用的 swarm 要格外留足时间——sequential_workflow_as_tool.py 把两个 Agent 串成SequentialWorkflow后特意设置了timeout300并在注释里说明两次模型调用是串行的给它留点空间。4.2 附加普通函数extra_toolsextra_tools可以在一台服务器上、主工具之外再暴露一批纯函数工具源码第 561-562 行直接把每个函数注册为server.tool。env_keys_and_extra_tools.py 演示了 Agent 两个辅助函数获取 UTC 时间、统计词数共存def utc_now() - str: The current UTC time as an ISO 8601 string. return datetime.datetime.now(datetime.timezone.utc).isoformat() deployer MCPDeployer( editor, api_key_envMCP_DEPLOYER_KEYS, # 例如 MCP_DEPLOYER_KEYSkey-a,key-b extra_tools[utc_now, word_count], port8004, )客户端侧会看到三个工具editor、utc_now、word_count。注意纯函数必须写 docstring因为函数的签名与 docstring 会成为 MCP 工具的 schema 与描述。五、三种传输协议Streamable HTTP、SSE 与 stdiotransport参数接受三个取值源码第 284-288 行会校验非法值streamable-http默认、sse、stdio。鉴权只对两种 HTTP 传输生效。5.1 Streamable HTTP默认默认传输端点路径默认为/mcp。除了鉴权还有两个针对该传输的开关json_responseTrue返回普通 JSON 而不是事件流。当目标只是普通函数、没有流式输出需求时这是更轻量的选择stateless_httpTrue默认开启不维护按会话session的状态适合部署在负载均衡之后源码第 249-253 行、第 625-631 行。plain_function_json_response.py 是无 LLM方案的完整样板目标是一个纯analyse函数统计字符数、词数、最长单词配合json_responseTrue、timeout5然后用MCPManager直接作为客户端调用deployer MCPDeployer( analyse, api_keys[stats-key], portfree_port(), json_responseTrue, # 返回纯 JSON而非事件流 timeout5, ) with deployer: client MCPManager(mcp_urldeployer.url, api_keystats-key) print(tools:, client.list_tool_names()) result client.call_tool(analyse, {task: the quick brown fox jumps over the lazy dog}) print(result[result])这证明部署 MCP 服务根本不需要任何 LLM API Key纯函数同样可以成为标准 MCP 工具。MCPManager.list_tool_names()与call_tool(name, arguments)都是它的便捷同步接口见 swarms/tools/mcp_manager.py 第 894-898 行与第 1094-1098 行。5.2 SSEtransportsse时端点路径自动变为/sse源码第 296 行通过 SSE 事件流传输。sse_transport.py 只需一行改动deployer MCPDeployer( translator, transportsse, api_keys[sk-local-dev], port8005, )5.3 stdiotransportstdio不启动任何 HTTP 服务而是由 MCP 宿主host通过标准输入/输出拉起——典型场景是桌面客户端配置一条指向该脚本的命令。由于 stdio 没有 HTTP 头鉴权设置会被忽略源码第 730-737 行会打印警告并直接server.run(stdio)因此通常配合allow_anonymousTrue宿主进程本身就是边界。stdio_transport.pydeployer MCPDeployer( assistant, transportstdio, allow_anonymousTrue, ) deployer.run()注意start()只支持 HTTP 传输stdio 会直接抛ValueError源码第 755-756 行而且 stdio 也没有 ASGI appbuild_app会拒绝。5.4 绑定地址与 DNS 重绑定防护host默认127.0.0.1。当绑定到非 localhost 地址时源码_transport_security第 600-611 行会关闭 MCP 传输层默认的 DNS 重绑定防护该防护只放行 localhost 的 Host 头否则外部绑定会被全部拒绝——这一细节说明跨主机部署时请自行确认网络边界与防火墙策略因为此时防护被有意关闭了。六、参数速查表基于 swarms/structs/mcp_deployer.py 构造函数第 257-331 行将全部参数整理如下参数默认值说明targets必填单个目标、目标 list或「工具名 → 目标」的 dictNone直接抛ValueErrorname首个工具名向 MCP 客户端广播的服务器名description目标的 description/docstring单目标场景下的工具描述tool_name目标的 snake_case 名单目标场景下的工具名host127.0.0.1绑定地址port8000绑定端口transportstreamable-httpstreamable-http/sse/stdiopath/mcpSSE 为/sseMCP 端点 URL 路径api_keysNone静态 Key 列表常量时间比较api_key_envNone环境变量名值为逗号分隔的多个 Key构造时读取api_key_headerx-api-key专用 Key 头Authorization: Bearer始终可用authNone自定义校验回调同步/异步返回 bool/dict/None优先级最高token_verifierNonemcp包的TokenVerifier实现required_scopesNone通过校验的 token 必须具备的作用域allow_anonymousFalse显式匿名未配置任何凭据时构造函数拒绝构建public_paths(/health,)免鉴权路径extra_toolsNone附加的普通函数必须有 docstringtimeoutNone单次工具调用的超时秒数json_responseFalseStreamable HTTP返回普通 JSON 而非事件流stateless_httpTrueStreamable HTTP不维护会话状态适合负载均衡场景verboseFalse记录每一次被放行的调用show_bannerTrue是否打印启动横幅swarms 的紫色外星人图案七、源码导出与快速部署入口MCPDeployer已从 swarms 顶层包导出见 swarms/structs/init.py 第 56 行from swarms.structs.mcp_deployer import MCPDeployer, deploy_as_mcp因此可以直接from swarms import Agent, MCPDeployer使用。客户端侧的工具同理MCPManager从 swarms/tools/init.py 导出。源码还提供了一个便捷函数deploy_as_mcp(target, **kwargs)swarms/structs/mcp_deployer.py 第 805-809 行构建MCPDeployer并立即阻塞运行适合在一次性脚本里快速起服务。八、典型落地组合从部署到消费的完整链路综合上面的全部要点一个多租户 Agent 服务 后台 Agent 消费的完整链路可以这样组织部署端用custom_auth_per_tenant.py的异步auth回调实现租户隔离或按 multiple_agents_one_server.py 把研究、写作、评审等多个 Agent 与辅助函数装进同一台服务器各自成为独立工具消费端在另一个Agent中通过mcp_urlMCPConnection(urlhttp://host:port/mcp, api_key..., headers{x-tenant: acme})连接——MCPConnection支持headers字段swarms/schemas/mcp_schemas.py 第 156-158 行可以让 LLM 自动把远端工具当作函数调用面向确定性调用场景则直接用MCPManager.call_tool(name, arguments)运维用公开的/health做健康探测用stateless_httpTrue部署到负载均衡之后用timeout约束单次调用时长必要时用api_key_env从环境变量注入 Key 而无需改动代码。关于模型选择所有示例均以普通 LiteLLM 字符串命名模型如gpt-5.4、claude-sonnet-4-6、claude-haiku-4-5-20251001、openrouter/moonshotai/kimi-k3、openrouter/z-ai/glm-5.3README 明确提示可以随意换成你拥有 Key 的任何厂商模型。唯一的例外是纯函数目标plain_function_json_response.py——它不需要任何 LLM Key因为根本没有模型参与。说明部署 MCP 服务器会对外开放一个可执行 Agent/swarm 的端口请务必为生产环境配置至少一种鉴权方式自定义auth、token_verifier或静态 Key切勿直接使用allow_anonymousTrue暴露在不可信网络上。【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 swarms 框架 MCPDeployer 实战指南：将 Agent 与任意 Swarm 一键部署为带鉴权的 MCP 服务器
swarms 框架 MCPDeployer 实战指南将 Agent 与任意 Swarm 一键部署为带鉴权的 MCP 服务器【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms导读MCPDeployer是 swarms 框架中负责服务化的核心组件它能把一个Agent、任何具备run()方法的 swarm例如SequentialWorkflow、SwarmRouter甚至一个普通 Python 函数包装成符合 MCPModel Context Protocol规范的服务器并在每个 HTTP 请求进入传输层之前插入一层鉴权。本文以 examples/mcp/mcp_deployer/README.md 为主线结合 swarms/structs/mcp_deployer.py 源码与 11 个示例文件系统讲解部署方式、鉴权优先级、生命周期管理、三种传输协议以及如何在另一个 Agent 中通过MCPConnection/MCPManager消费这些服务。读完你可以在几分钟内把任意 Agent 变成带 API Key、租户隔离或 OAuth 风格 Token 校验的远程 MCP 工具。一、MCPDeployer 是什么一个目标 一个 MCP 工具MCPDeployer的核心模型非常简单每一个被部署的目标target都会成为服务器上的一个 MCP 工具tool工具名取自目标名输入 schema 统一为(task, img)——task是任务字符串img是可选图片参数默认为None。可被部署的目标需要满足_is_servable的判定见 swarms/structs/mcp_deployer.py 第 96-98 行目标要么可调用callable要么具备可调用的run方法。因此以下对象都可以直接部署单个Agent任何带run(task, ...)方法的 swarm 结构SequentialWorkflow、SwarmRouter、HeavySwarm等接收任务字符串的普通 Python 函数。1.1 最简部署一个 AgentREADME 给出的最小可用代码只有三行from swarms import Agent, MCPDeployer agent Agent(agent_nameResearcher, model_namegpt-5.4, max_loops1) MCPDeployer(agent, api_keys[sk-local-dev], port8000).run()运行后服务器会在127.0.0.1:8000上监听MCP 端点默认为/mcp。另一个 Agent 只需这样连接Agent(mcp_urlMCPConnection(urlhttp://127.0.0.1:8000/mcp, api_keysk-local-dev))1.2 工具名的自动推导如果不显式指定工具名MCPDeployer会按照_default_tool_nameswarms/structs/mcp_deployer.py 第 100-109 行自动生成一个 snake_case 名称取值优先级为agent_name→name→__name__→ 类型名。例如名为Researcher的 Agent 会暴露为工具researcher。工具描述description同样自动推导_default_description第 112-120 行优先取agent_description或description属性否则取 docstring 首行最后兜底为Run tool_name on a task.。1.3 单服务器承载多个目标把目标传成list或dict即可在一个服务器上同时暴露多个工具。dict 的 key 就是工具名见 swarms/structs/mcp_deployer.py 第 342-360 行MCPDeployer( {research: researcher, write: writer, review: pipeline}, api_keys[sk-local-dev], ).run()注意传入 dict 或 list 时不能再传tool_name或description那仅适用于单目标场景否则构造函数会抛出ValueError。服务器启动后还可以通过add_tool(target, name, description)在run()/start()之前动态注册更多工具。1.4 部署目标的内部调用链从源码_invokeswarms/structs/mcp_deployer.py 第 525-538 行可以看出每次工具调用的执行逻辑若目标有run方法且客户端传了img先尝试run(task, imgimg)遇到TypeError则回退有run方法则调用run(task)否则直接把目标当函数调用target(task)。返回值统一经过_to_text第 123-132 行渲染为字符串字符串原样返回其他对象dict、list 等用json.dumps(result, indent2, defaultstr)序列化。也就是说函数目标返回任意可序列化对象都可以客户端拿到的永远是结构化文本结果。二、示例全景11 个可复制的实战脚本examples/mcp/mcp_deployer 目录下共 11 个示例覆盖了不同的目标类型、鉴权方式与传输协议下面的总表来自 README 并已核对各脚本实际内容示例目标鉴权传输协议是否独立运行single_agent_api_key.py单个Agent静态api_keysstreamable HTTP持续服务直到被中断sequential_workflow_as_tool.pySequentialWorkflow静态api_keysstreamable HTTP持续服务直到被中断multiple_agents_one_server.py两个 Agent、一个SequentialWorkflow、两个函数各自独立成工具静态api_keysstreamable HTTP持续服务直到被中断custom_auth_per_tenant.py单个Agent异步auth回调读取x-tenantstreamable HTTP持续服务直到被中断owner_key_or_tenant_auth.py单个Agent同步auth回调环境变量中的 owner key或白名单x-tenantstreamable HTTP持续服务直到被中断token_verifier_with_scopes.py单个AgentTokenVerifier协议 required_scopesstreamable HTTP持续服务直到被中断env_keys_and_extra_tools.py单个Agent外加两个普通函数api_key_envstreamable HTTP持续服务直到被中断background_server_and_client_agent.py单个Agent静态api_keysstreamable HTTP是后台服务、调用、自动停止plain_function_json_response.py纯函数无 LLM静态api_keysstreamable HTTPJSON 响应是无需任何 LLM Keysse_transport.py单个Agent静态api_keysSSE持续服务直到被中断stdio_transport.py单个Agent无宿主即边界stdio由 MCP 宿主host拉起所有示例的模型名都使用普通的 LiteLLM 字符串如gpt-5.4、claude-sonnet-4-6、openrouter/moonshotai/kimi-k3你可以随意替换成任何你拥有 API Key 的厂商模型不需要改动其他任何代码。三、鉴权体系优先级明确的四层防护MCPDeployer最与众不同的能力是把鉴权层放在 MCP 传输层之前。从源码实现看_AuthMiddlewareswarms/structs/mcp_deployer.py 第 166-203 行是一个 ASGI 中间件它对每个 HTTP 请求先调用deployer.authenticate(headers)只有返回AuthResult的请求才会被放行到内部的 MCP 应用被拒绝的请求返回401并附带WWW-Authenticate: Bearer响应头。鉴权检查的顺序authenticate方法第 459-519 行与 README 的Auth, in order of precedence完全一致3.1 第一优先级authcallable(credential, headers)你自己的校验函数同步或异步均可签名是(credential, headers) - bool | dict | None返回真值truthy→ 放行返回dict→ 放行并把这个 dict 作为该请求的 claims声明可供后续工具调用上下文使用返回假值或抛出异常 → 拒绝。从源码第 481-493 行可以看到auth返回的 dict 会被包装成AuthResult其中subject取claims.get(subject) or claims.get(sub)scopes取claims.get(scopes, [])。owner_key_or_tenant_auth.py 展示了最典型的多重判断凭据等于环境变量MCP_DEPLOYER_KEY时以owner身份放行或者x-tenant头在白名单{acme, globex}内时按租户放行否则拒绝def is_allowed(credential, headers): if credential and credential os.getenv(MCP_DEPLOYER_KEY): return {subject: owner, scopes: [run]} tenant headers.get(x-tenant) if tenant in {acme, globex}: return {subject: tenant, scopes: [run]} return False deployer MCPDeployer(researcher, port8000, authis_allowed, timeout120, verboseTrue)custom_auth_per_tenant.py 则展示了异步版本每个租户持有独立 Key凭据与TENANT_KEYS[tenant]匹配后才放行并把{subject: tenant, scopes: [run], tenant: tenant}作为 claims 写入请求。这就是标准的一租户一密钥多租户隔离方案。3.2 第二优先级token_verifierOAuth 风格 Token当未配置auth时可以传入mcp包定义的TokenVerifier协议实现。authenticate会调用await token_verifier.verify_token(credential)并强制校验两点源码第 498-515 行过期时间token.expires_at若小于当前时间则拒绝作用域required_scopes中的每一项都必须出现在 token 的scopes里缺一即拒绝。token_verifier_with_scopes.py 给出了完整示例签发两个 tokentok-admin带[agent:run, agent:admin]且一小时内有效tok-reader只有[agent:read]注释明确说明它会因缺少agent:run而被 401 拒绝class StaticTokenVerifier: async def verify_token(self, token: str): return ISSUED_TOKENS.get(token) deployer MCPDeployer( summariser, token_verifierStaticTokenVerifier(), required_scopes[agent:run], port8003, )3.3 第三优先级api_keys[...]与api_key_envVAR静态 Keyapi_keys直接传入的静态 Key 列表api_key_env环境变量名变量值为逗号分隔的多个 Key在构造时读取_collect_api_keys第 441-451 行因此可以写成MCP_DEPLOYER_KEYSkey-a,key-b。静态 Key 使用hmac.compare_digest进行常量时间比较第 453-457 行避免时序侧信道攻击。需要留意环境变量读取发生在MCPDeployer(...)构造那一刻运行时再修改环境变量不会生效。3.4 第四优先级allow_anonymousTrue显式匿名匿名访问必须显式开启。源码第 314-323 行有一条强约束如果allow_anonymous为假且auth、token_verifier、api_keys全部未配置构造函数直接抛出ValueError杜绝裸奔上线。3.5 客户端如何携带凭据README 明确说明两种方式专用头x-api-key可用api_key_header参数改名标准Authorization: Bearer token这正是MCPManager默认发送的方式。credential_from_headers源码第 135-163 行的解析逻辑是先读api_key_header默认x-api-key带Bearer前缀时自动剥掉读不到再退回Authorization: Bearer。客户端侧的MCPConnection定义于 swarms/schemas/mcp_schemas.py 第 111-159 行默认api_key_headerAuthorization、api_key_prefixBearer与服务器端默认完全对齐。3.6 公共路径与健康检查public_paths默认为(/health,)第 65 行即/health永远公开。_build_server里注册的健康检查路由第 566-576 行会返回status、服务器name、当前tool、全部tools列表与transport类型非常适合接入负载均衡器或 K8s 探针。四、生命周期管理阻塞、后台线程与上下文管理器MCPDeployer提供三种运行方式README 的 Lifecycle 一节run()阻塞式持续服务直到 CtrlC 中断内部走uvicorn.run见源码第 726-746 行start()/stop()把 uvicorn 服务器放进后台线程运行start()会等待 socket 打开后返回wait参数控制最长等待秒数默认 10 秒超时会抛RuntimeError见第 748-787 行with MCPDeployer(...) as d:上下文管理器进入时start()、退出时自动stop()第 798-802 行。background_server_and_client_agent.py 把后台服务 客户端调用 自动停止完整串了起来with MCPDeployer(poet, api_keys[demo-key], portfree_port()) as server: print(fserving {server.tool_name} at {server.url}) client Agent( agent_nameClient, system_promptUse the poet tool to get a poem, then return it verbatim., model_nameclaude-haiku-4-5-20251001, mcp_urlMCPConnection(urlserver.url, api_keydemo-key), max_loops2, print_onFalse, output_typefinal, ) client.run(Get a poem about lighthouses.) print(poet.short_memory.get_final_message_content()) # 离开 with 块即停止服务器这里还展示了两个实用细节server.url属性会拼出完整的端点地址http://host:port/path源码第 596-598 行服务端 Agent 的结果可以通过short_memory.get_final_message_content()从内存中取回。4.1 单次工具调用的超时控制timeout参数以秒为单位约束单次工具调用的最大执行时间。源码第 547-551 行用anyio.fail_after(self.timeout)包裹同步调用通过anyio.to_thread.run_sync丢到线程池执行。对串联多次模型调用的 swarm 要格外留足时间——sequential_workflow_as_tool.py 把两个 Agent 串成SequentialWorkflow后特意设置了timeout300并在注释里说明两次模型调用是串行的给它留点空间。4.2 附加普通函数extra_toolsextra_tools可以在一台服务器上、主工具之外再暴露一批纯函数工具源码第 561-562 行直接把每个函数注册为server.tool。env_keys_and_extra_tools.py 演示了 Agent 两个辅助函数获取 UTC 时间、统计词数共存def utc_now() - str: The current UTC time as an ISO 8601 string. return datetime.datetime.now(datetime.timezone.utc).isoformat() deployer MCPDeployer( editor, api_key_envMCP_DEPLOYER_KEYS, # 例如 MCP_DEPLOYER_KEYSkey-a,key-b extra_tools[utc_now, word_count], port8004, )客户端侧会看到三个工具editor、utc_now、word_count。注意纯函数必须写 docstring因为函数的签名与 docstring 会成为 MCP 工具的 schema 与描述。五、三种传输协议Streamable HTTP、SSE 与 stdiotransport参数接受三个取值源码第 284-288 行会校验非法值streamable-http默认、sse、stdio。鉴权只对两种 HTTP 传输生效。5.1 Streamable HTTP默认默认传输端点路径默认为/mcp。除了鉴权还有两个针对该传输的开关json_responseTrue返回普通 JSON 而不是事件流。当目标只是普通函数、没有流式输出需求时这是更轻量的选择stateless_httpTrue默认开启不维护按会话session的状态适合部署在负载均衡之后源码第 249-253 行、第 625-631 行。plain_function_json_response.py 是无 LLM方案的完整样板目标是一个纯analyse函数统计字符数、词数、最长单词配合json_responseTrue、timeout5然后用MCPManager直接作为客户端调用deployer MCPDeployer( analyse, api_keys[stats-key], portfree_port(), json_responseTrue, # 返回纯 JSON而非事件流 timeout5, ) with deployer: client MCPManager(mcp_urldeployer.url, api_keystats-key) print(tools:, client.list_tool_names()) result client.call_tool(analyse, {task: the quick brown fox jumps over the lazy dog}) print(result[result])这证明部署 MCP 服务根本不需要任何 LLM API Key纯函数同样可以成为标准 MCP 工具。MCPManager.list_tool_names()与call_tool(name, arguments)都是它的便捷同步接口见 swarms/tools/mcp_manager.py 第 894-898 行与第 1094-1098 行。5.2 SSEtransportsse时端点路径自动变为/sse源码第 296 行通过 SSE 事件流传输。sse_transport.py 只需一行改动deployer MCPDeployer( translator, transportsse, api_keys[sk-local-dev], port8005, )5.3 stdiotransportstdio不启动任何 HTTP 服务而是由 MCP 宿主host通过标准输入/输出拉起——典型场景是桌面客户端配置一条指向该脚本的命令。由于 stdio 没有 HTTP 头鉴权设置会被忽略源码第 730-737 行会打印警告并直接server.run(stdio)因此通常配合allow_anonymousTrue宿主进程本身就是边界。stdio_transport.pydeployer MCPDeployer( assistant, transportstdio, allow_anonymousTrue, ) deployer.run()注意start()只支持 HTTP 传输stdio 会直接抛ValueError源码第 755-756 行而且 stdio 也没有 ASGI appbuild_app会拒绝。5.4 绑定地址与 DNS 重绑定防护host默认127.0.0.1。当绑定到非 localhost 地址时源码_transport_security第 600-611 行会关闭 MCP 传输层默认的 DNS 重绑定防护该防护只放行 localhost 的 Host 头否则外部绑定会被全部拒绝——这一细节说明跨主机部署时请自行确认网络边界与防火墙策略因为此时防护被有意关闭了。六、参数速查表基于 swarms/structs/mcp_deployer.py 构造函数第 257-331 行将全部参数整理如下参数默认值说明targets必填单个目标、目标 list或「工具名 → 目标」的 dictNone直接抛ValueErrorname首个工具名向 MCP 客户端广播的服务器名description目标的 description/docstring单目标场景下的工具描述tool_name目标的 snake_case 名单目标场景下的工具名host127.0.0.1绑定地址port8000绑定端口transportstreamable-httpstreamable-http/sse/stdiopath/mcpSSE 为/sseMCP 端点 URL 路径api_keysNone静态 Key 列表常量时间比较api_key_envNone环境变量名值为逗号分隔的多个 Key构造时读取api_key_headerx-api-key专用 Key 头Authorization: Bearer始终可用authNone自定义校验回调同步/异步返回 bool/dict/None优先级最高token_verifierNonemcp包的TokenVerifier实现required_scopesNone通过校验的 token 必须具备的作用域allow_anonymousFalse显式匿名未配置任何凭据时构造函数拒绝构建public_paths(/health,)免鉴权路径extra_toolsNone附加的普通函数必须有 docstringtimeoutNone单次工具调用的超时秒数json_responseFalseStreamable HTTP返回普通 JSON 而非事件流stateless_httpTrueStreamable HTTP不维护会话状态适合负载均衡场景verboseFalse记录每一次被放行的调用show_bannerTrue是否打印启动横幅swarms 的紫色外星人图案七、源码导出与快速部署入口MCPDeployer已从 swarms 顶层包导出见 swarms/structs/init.py 第 56 行from swarms.structs.mcp_deployer import MCPDeployer, deploy_as_mcp因此可以直接from swarms import Agent, MCPDeployer使用。客户端侧的工具同理MCPManager从 swarms/tools/init.py 导出。源码还提供了一个便捷函数deploy_as_mcp(target, **kwargs)swarms/structs/mcp_deployer.py 第 805-809 行构建MCPDeployer并立即阻塞运行适合在一次性脚本里快速起服务。八、典型落地组合从部署到消费的完整链路综合上面的全部要点一个多租户 Agent 服务 后台 Agent 消费的完整链路可以这样组织部署端用custom_auth_per_tenant.py的异步auth回调实现租户隔离或按 multiple_agents_one_server.py 把研究、写作、评审等多个 Agent 与辅助函数装进同一台服务器各自成为独立工具消费端在另一个Agent中通过mcp_urlMCPConnection(urlhttp://host:port/mcp, api_key..., headers{x-tenant: acme})连接——MCPConnection支持headers字段swarms/schemas/mcp_schemas.py 第 156-158 行可以让 LLM 自动把远端工具当作函数调用面向确定性调用场景则直接用MCPManager.call_tool(name, arguments)运维用公开的/health做健康探测用stateless_httpTrue部署到负载均衡之后用timeout约束单次调用时长必要时用api_key_env从环境变量注入 Key 而无需改动代码。关于模型选择所有示例均以普通 LiteLLM 字符串命名模型如gpt-5.4、claude-sonnet-4-6、claude-haiku-4-5-20251001、openrouter/moonshotai/kimi-k3、openrouter/z-ai/glm-5.3README 明确提示可以随意换成你拥有 Key 的任何厂商模型。唯一的例外是纯函数目标plain_function_json_response.py——它不需要任何 LLM Key因为根本没有模型参与。说明部署 MCP 服务器会对外开放一个可执行 Agent/swarm 的端口请务必为生产环境配置至少一种鉴权方式自定义auth、token_verifier或静态 Key切勿直接使用allow_anonymousTrue暴露在不可信网络上。【免费下载链接】swarmsThe Enterprise-Grade Multi-Agent Orchestration Framework. Website: https://swarms.ai项目地址: https://gitcode.com/GitHub_Trending/swar/swarms创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考