mcpField
S
swarmmesh

swarmmesh

本地服务
github.com/RudrenduPaul记忆与知识05 周前更新

关于 swarmmesh

SwarmMesh 为并行 AI Agent 提供共享上下文、记忆和状态协调服务,供不同进程及语言的 Agent 读写同一状态。它不负责调度任务,采用公开的 HTTP 协议,并提供 Python 与 Node 实现及 BM25 记忆检索。

主要功能

  1. 01通过 HTTP 协议共享上下文
  2. 02支持 Python 与 Node 实现互通
  3. 03使用 WebSocket 推送实时更新
  4. 04通过 BM25 对记忆进行关键词排序
  5. 05支持内存或 SQLite 存储
  6. 06提供 JSON 命令行输出

适用场景

  1. 01让并行 Agent 共享任务上下文
  2. 02避免 Agent 重复发现或覆盖变更
  3. 03跨 Python 和 Node Agent 查询记忆
  4. 04在状态变化时触发 Agent 响应

MCP 相关

提供 Python 和 Node 版本的 MCP 服务,暴露十个协调工具;服务将工具调用通过 HTTP 转发到已运行的 SwarmMesh 服务。

配置

本地服务的配置在各客户端中通用,复制后粘贴到客户端的 MCP 配置文件即可。

mcp.json · 通用配置
{
"mcpServers": {
"rudrendupaul-swarmmesh": {
"command": "uvx",
"args": [
"swarmmesh-cli"
]
}
}
}
启动命令uvx swarmmesh-cli

本地服务需要在你的电脑上安装运行环境(如 Node.js 18+)。各客户端的配置文件位置不同,请参考客户端文档。

安装包
pypiswarmmesh-cli0.1.3

工具

10

收录的工具列表 · 更新于 2 周前

参数类型必填说明
rolestring是
agent_idstring是
metadataany否

详细介绍

SwarmMesh

CI (Python) CI (Node) PyPI npm License: MIT

安装 • 快速入门 • 功能 • 命令行参考 • 对比 • 常见问题

为并行工作的 AI Agent 群组共享上下文和记忆,采用一种 Python 和 Node 都能以相同方式使用的小型协议。

swarmmesh 演示:启动网格、注册 Agent、写入上下文和记忆,然后查询记忆

在同一任务上启动十个编码 Agent,它们却无法看到彼此发现的内容。一个 Agent 重新发现了另一个 Agent 已经修复的错误。两个 Agent 覆盖了同一个文件,因为双方都不知道对方改动过它。SwarmMesh 是一个小型服务,与现有的 Agent 框架并行运行,让每个 Agent 进程都能在任何支持 HTTP 的语言中发布上下文并搜索记忆。

它不是编排框架。它不安排任务、不定义 Agent 角色,也不在 Agent 之间分配工作。现有框架(或你自己的代码)继续负责这些工作。SwarmMesh 只回答一个问题:独立的 Agent 进程如何读写同一份共享状态。

安装

pip install swarmmesh-cli
# or
npm install -g swarmmesh-cli

两种方式都可以在你的 PATH 中提供 swarmmesh 命令。

看看它如何运行

这是一段真实的终端会话,而不是模拟:由 Python 运行的网格、向其中写入数据的 Node Agent,以及读取 Node Agent 所写内容的 Python Agent。两种不同的语言,一个共享网格。

# Terminal 1: start a mesh (Python implementation, but either works)
$ swarmmesh serve --port 8420
INFO: Uvicorn running on http://127.0.0.1:8420

# Terminal 2: a Node agent joins and writes
$ swarmmesh agent register node-agent-1 researcher --port 8420 --json
{ "agent_id": "node-agent-1", "role": "researcher", ... }

$ swarmmesh context set interop-demo status '"investigating flaky test"' \
    --agent-id node-agent-1 --port 8420 --json
{ "namespace": "interop-demo", "key": "status", "value": "investigating flaky test", ... }

$ swarmmesh memory write interop-demo \
    "found a race condition in the retry loop" --agent-id node-agent-1 --port 8420 --json
{ "namespace": "interop-demo", "text": "found a race condition in the retry loop", ... }

# Terminal 3: a Python agent joins the same mesh and reads it back
$ swarmmesh context get interop-demo status --port 8420 --json
{ "value": "investigating flaky test", "updated_by": "node-agent-1", ... }

$ swarmmesh memory query interop-demo "race condition" --port 8420 --json
{ "results": [{ "entry": { "text": "found a race condition in the retry loop" }, "score": 0.575 }] }

编写此 README 时,上述每条命令都针对两个 CLI 重新真实运行过:Node CLI 在 Python 托管的网格上注册了 Agent,并写入上下文和记忆;Python CLI 在同一次运行中通过真实 HTTP API 直接读回了这些内容,并精确复现了上方的分数(0.575)。没有共享文件系统,没有共享进程,也没有转换层,只有协议。

快速入门

# Start a mesh (in-memory by default; add --persist ./mesh.db for SQLite storage)
swarmmesh serve --host 127.0.0.1 --port 8420

# From another terminal: register an agent
swarmmesh agent register agent-1 researcher

# Publish and read shared context
swarmmesh context set my-run phase '"planning"' --agent-id agent-1
swarmmesh context get my-run phase

# Write and search shared memory
swarmmesh memory write my-run "found a race condition in the retry loop" --agent-id agent-1
swarmmesh memory query my-run "race condition"

# Check what's on the mesh
swarmmesh status --json

编写此 README 时,我们端到端运行了这一整套命令,从开始到结束只用了几秒,并且使用的是从 PyPI 安装的真实 swarmmesh-cli 包。

如果要从源代码构建,而不是从软件包注册表安装:

# Python
git clone https://github.com/RudrenduPaul/swarmmesh.git
cd swarmmesh
pip install -e python/

# Node
cd swarmmesh/node
npm install
npm run build
npm link

功能

  • 有据可查的线协议。 docs/protocol.md 说明了每个 HTTP 端点和 WebSocket 事件,因此任何能够使用 HTTP 和 JSON 通信的进程都可以加入网格。两个官方 CLI 是便捷的客户端,但并非唯一可用的客户端。
  • 两个相互独立且可互操作的实现。 Python (PyPI 上的 swarmmesh-cli,FastAPI + Typer,74 个测试,语句 覆盖率为 91%)和 Node(npm 上的 swarmmesh-cli,Express + commander,65 个测试, 语句覆盖率为 91.64%)以相同方式实现了该协议。每个 软件包自己的测试套件都会在 CI 中独立运行;跨语言互操作 (Node 客户端连接 Python 托管的服务,反之亦然)在上方的“查看实际效果”一节中展示过, 并且曾针对两个真实软件包手动重新运行,但目前 CI 中没有自动化的跨语言测试。
  • 通过 WebSocket 实时更新。 /v1/events 会推送 context.updated、context.deleted、memory.written、 agent.registered 和 agent.deregistered 帧,让 Agent 可以在 另一个 Agent 更改共享状态时立即响应,而不必轮询。
  • 如实呈现的记忆搜索。 记忆查询使用 Okapi BM25 关键词排序: 在本地计算真实的词频评分,无需额外依赖,也不会发起网络调用。它不是语义搜索或嵌入搜索。RankingBackend 接口是有文档说明的扩展点,可供你接入自己的基于嵌入的评分器;SwarmMesh 不附带此类评分器。
  • 可插拔存储。 默认使用内存存储(仅在进程生命周期内有效),也可以使用 --persist <path> 启用基于 SQLite 的存储,数据可在重启后保留。
  • 默认面向 Agent。 两个 CLI 的每个子命令都支持 --json,以输出结构化、可由脚本解析的内容;两者还都提供 swarmmesh mcp 子命令,可通过 stdio 启动 MCP 服务,让支持 MCP 的 Agent(Claude 或其他 Agent)无需调用 Shell,即可将 SwarmMesh 用作一组工具。
  • 刻意保持较小的信任边界。 默认情况下,两个服务都绑定到 127.0.0.1,而不是 0.0.0.0。v1 中没有身份验证。 请参阅安全性。

以下数字是实测值,而非估算值。在运行本文撰写时所用机器上, 对本地 Python 服务连续发送 50 个 PUT /v1/context/{namespace}/{key} 请求,平均每个请求的往返时间为 0.8ms (50 个请求总计 40ms)。 这并非严谨的基准测试,包含了每次请求启动 curl 进程的开销, 且会因机器而异;不过这是实际运行得出的真实数字,并非猜测。你可以自行复现:

for i in $(seq 1 50); do curl -s -o /dev/null -w "%{time_total}\n" \
  -X PUT "http://127.0.0.1:8420/v1/context/bench/key$i" \
  -H "Content-Type: application/json" -d "{\"value\":\"v$i\",\"agent_id\":\"bench\"}"; done

CLI 参考

两个 CLI 提供相同的命令树。标志名称略有不同 (Python 使用 Typer 的 --flag <value> 风格,Node 使用 commander 的风格), 但命令及其行为完全相同。以下输出 摘录自对每个构建好的 CLI 运行 --help 的结果。

swarmmesh --help 和 swarmmesh agent --help 的输出

swarmmesh serve [--host HOST] [--port PORT] [--persist PATH]
    Start a SwarmMesh coordination server.

swarmmesh status [--host HOST] [--port PORT] [--json]
    Show a mesh status snapshot (agent count, namespaces, entry counts, uptime).

swarmmesh mcp [--host HOST] [--port PORT]
    Start an MCP server over stdio, proxying tool calls to a running mesh.

swarmmesh agent register <agent_id> <role> [--metadata JSON] [--host HOST] [--port PORT] [--json]
swarmmesh agent list [--host HOST] [--port PORT] [--json]
swarmmesh agent deregister <agent_id> [--host HOST] [--port PORT] [--json]

swarmmesh context set <namespace> <key> <value> [--agent-id ID] [--ttl SECONDS] [--host HOST] [--port PORT] [--json]
swarmmesh context get <namespace> <key> [--host HOST] [--port PORT] [--json]
swarmmesh context list <namespace> [--host HOST] [--port PORT] [--json]
swarmmesh context delete <namespace> <key> [--host HOST] [--port PORT] [--json]

swarmmesh memory write <namespace> <text> [--agent-id ID] [--metadata JSON] [--id ID] [--host HOST] [--port PORT] [--json]
swarmmesh memory query <namespace> <query> [--top-k N] [--host HOST] [--port PORT] [--json]

注册 Agent,然后运行 swarmmesh status --json,并在运行中的网格上设置和列出上下文

context set 会将 <value> 解析为 JSON;如果不是有效的 JSON,则回退为普通字符串。context set ns key '"planning"' 会存储字符串 planning。通过相同的字符串回退机制,context set ns key planning(不加引号)也会存储该字符串。

MCP 服务

SwarmMesh 的 Python 和 Node 包均内置了模型上下文协议(MCP)服务,因此支持 MCP 的 Agent(Claude Desktop、Claude Code 或任何其他 MCP 客户端)可以将 SwarmMesh 作为一组工具调用,而无需通过 Shell 调用 CLI。MCP 服务不会重新实现该协议;它会通过 HTTP 将每次工具调用转发给已在运行的 swarmmesh serve 进程。

# 1. Start a mesh
swarmmesh serve --host 127.0.0.1 --port 8420

# 2. In another terminal (or from an MCP client), start the MCP server
#    (stdio transport) pointed at that mesh:
swarmmesh mcp --host 127.0.0.1 --port 8420

两个包默认都包含 mcp 支持(它是核心依赖,而非可选扩展),因此只需执行 pip install swarmmesh-cli 或 npm install -g swarmmesh-cli 即可。

Claude Desktop 配置(claude_desktop_config.json):

{
  "mcpServers": {
    "swarmmesh": {
      "command": "swarmmesh",
      "args": ["mcp", "--host", "127.0.0.1", "--port", "8420"]
    }
  }
}

Python 和 Node MCP 服务都提供相同的 10 个工具,与上文的 SwarmMeshClient 方法相对应:

工具功能调用示例
register_agent在网格中注册 Agent。register_agent(agent_id="agent-1", role="researcher")
deregister_agent从网格中注销 Agent。此操作幂等。deregister_agent(agent_id="agent-1")
list_agents列出当前在网格中注册的 Agent。list_agents()
publish_context在命名空间中发布(创建或覆盖)一个上下文值。publish_context(namespace="my-run", key="phase", value="planning", agent_id="agent-1")
get_context读取单个上下文值。get_context(namespace="my-run", key="phase")
list_context列出命名空间中所有有效(未过期)的上下文条目。list_context(namespace="my-run")
delete_context删除一个上下文值。delete_context(namespace="my-run", key="phase")
write_memory写入一条记忆,供群集中的其他 Agent 日后查找。write_memory(namespace="my-run", text="found a race condition in the retry loop", agent_id="agent-1")
query_memory按 BM25 关键词排名查询命名空间中的记忆条目(非语义搜索)。query_memory(namespace="my-run", query="race condition")
get_status获取网格状态快照(Agent 数量、命名空间、条目数量、运行时间)。get_status()

库 API 参考

两个包都导出了类型化客户端,因此你可以直接在自己的 Agent 代码中调用网格,而无需通过 Shell 调用 CLI。以下签名直接从源代码中提取,并非凭记忆编写。

Python(swarmmesh_cli.client.SwarmMeshClient):

class SwarmMeshClient:
    def __init__(self, base_url: str = DEFAULT_BASE_URL, timeout: float = 10.0) -> None: ...
    async def register_agent(self, agent_id: str, role: str, metadata: dict | None = None) -> dict: ...
    async def deregister_agent(self, agent_id: str) -> None: ...
    async def list_agents(self) -> dict: ...
    async def publish_context(self, namespace: str, key: str, value, agent_id: str, ttl_seconds: int | None = None) -> dict: ...
    async def get_context(self, namespace: str, key: str) -> dict: ...
    async def list_context(self, namespace: str) -> dict: ...
    async def delete_context(self, namespace: str, key: str) -> None: ...
    async def write_memory(self, namespace: str, text: str, agent_id: str, metadata: dict | None = None) -> dict: ...
    async def query_memory(self, namespace: str, query: str, top_k: int = 10) -> dict: ...
    async def get_status(self) -> dict: ...

Node / TypeScript(来自 swarmmesh-cli 的 SwarmMeshClient):

class SwarmMeshClient {
  constructor(options?: SwarmMeshClientOptions);
  registerAgent(agentId: string, role: string, metadata?: Record<string, JsonValue>): Promise<Agent>;
  deregisterAgent(agentId: string): Promise<void>;
  listAgents(): Promise<Agent[]>;
  publishContext(namespace: string, key: string, value: JsonValue, agentId: string, ttlSeconds?: number): Promise<ContextEntry>;
  getContext(namespace: string, key: string): Promise<ContextEntry | null>;
  listContext(namespace: string): Promise<ContextEntry[]>;
  deleteContext(namespace: string, key: string): Promise<void>;
  writeMemory(namespace: string, text: string, agentId: string, metadata?: Record<string, JsonValue>): Promise<MemoryEntry>;
  queryMemory(namespace: string, query: string, topK?: number): Promise<MemoryQueryResult[]>;
  getStatus(): Promise<StatusSnapshot>;
}

SwarmMesh 协议

完整规范见 docs/protocol.md。简而言之:“网格”是一个正在运行的 swarmmesh serve 进程。Agent 是独立进程(编码 Agent、研究 Agent、子进程工作线程,或任何能够发起 HTTP 请求的程序),它们向网格注册,然后通过网格读写带命名空间的共享上下文和记忆。

将其写成协议而非仅发布一个库的意义在于,两个官方 CLI 并非唯一有效的客户端。使用 swarmmesh_cli.client.SwarmMeshClient 的 Python Agent、使用 swarmmesh-cli 中 SwarmMeshClient 的 Node Agent,以及使用不支持这两个包的语言编写的第三个 Agent,都可以向同一个网格注册并查看彼此的上下文和记忆,因为它们都只是调用相同的、已记录的 HTTP 端点,并且可以选择订阅相同的 WebSocket 事件流。互操作不依赖共享运行时、共享进程或共享文件系统。

SwarmMesh 的对比

没有其他项目完全实现了 SwarmMesh 的功能,因此这不是一张可以直接比较的表格。列出这些内容,是为了如实说明两个真正可比的多 Agent 项目实际提供了什么,以及 SwarmMesh 实际提供了什么;这些信息均直接对照它们的 README 和源代码核实,而非根据项目名称推测。它们都比 SwarmMesh 历史更久、规模更大、成熟度更高;SwarmMesh 目前在 GitHub 上有 0 个 star,也暂无已知用户。

SwarmMeshkyegomez/swarmscompanion-inc/feynman
项目定位共享上下文/记忆协调层(基础设施,而非框架)多 Agent 编排框架带有本地工作台 UI 的 AI 研究 Agent
Star 数07,0248,447
主要语言Python + TypeScript(两种经过测试的实现)PythonTypeScript
许可证MITApache-2.0MIT
安装pip install swarmmesh-cli / npm install -g swarmmesh-clipip3 install -U swarmscurl -fsSL https://feynman.is/install | bash
用于共享上下文/记忆的跨语言协议文档有:docs/protocol.md,采用 HTTP + WebSocket;两个独立实现已通过手动验证,确认可以互操作(参见上文“查看实际效果”)并非主要功能。AOP 是一个真正的协议,用于将具名远程 Agent 部署并作为分布式服务调用,但其文档中的示例仅适用于 Python,且未指定语言无关的线上格式。它有一个 RedisConversation 后端作为示例工具,但没有将其作为跨语言协调功能来编写文档。未发现。feynman serve 会运行一个面向本地用户的工作台 UI。状态存储在 ~/.feynman/ 下的本地 SQLite 镜像中,并未通过有文档说明的 Agent 间 API 提供。
内置编排模式(顺序、分层、任务路由)按设计不提供。SwarmMesh 希望你自行选择编排器有,种类繁多。这是 swarms 的核心功能有一些,但仅用于其自身的研究工作流,并未作为通用 SDK 提供
记忆搜索关键词(BM25),明确不支持语义搜索不是项目重点不是项目重点

客观来说:swarms 具备真正深入的编排能力,也拥有庞大的社区,而 SwarmMesh 并不试图取而代之。feynman 是一款精致的终端用户研究工具,并非可嵌入其他产品的基础设施。SwarmMesh 的实际主张比两者都更狭窄:提供一种小巧、有文档说明的协议,让两种语言以相同方式进行通信。它的价值仅限于此,不多也不少。

SwarmMesh 是什么,以及它为何存在

多 Agent 配置日益意味着多个 Agent 进程并行处理同一个问题;有时它们使用相同的语言,有时不同,有时甚至由完全不同的工具启动。编排框架解决的是“每个 Agent 应该做什么、按什么顺序做”的问题。SwarmMesh 解决的是一个更狭窄但相关的问题:这些 Agent 启动后,如何互相告知各自的发现,而无需人工在终端之间传递消息,也无需担心 Agent 悄悄重复彼此的工作。

SwarmMesh 是基础设施,而非框架。它不关心是什么编排器启动了你的 Agent(如果有的话)。它提供精简的 HTTP + WebSocket 接口,用于共享上下文(结构化的键值状态,例如某次运行的当前阶段)和共享记忆(Agent 留给彼此的自由文本笔记,可按关键词搜索)。你可以像连接 Redis 实例一样,让 Agent 连接到 swarmmesh serve 进程,这样它们就能在同一个地方读写内容。

常见问题

这是 LangGraph / CrewAI / AutoGen / <我的编排框架> 的替代品吗? 不是。SwarmMesh 不会调度 Agent、定义工作流或决定接下来要做什么。它可以与您用于这些工作的工具一起运行,为其启动的 Agent 提供共享的上下文和记忆层。让您的编排器中的 Agent 连接到 swarmmesh serve 进程,其余工作仍由编排器负责。

这与 kyegomez/swarms 或 companion-inc/feynman 有何不同? 这两个项目规模更大、历史更久,解决的问题也不同。swarms 是一个编排框架:它决定运行哪些 Agent、按什么顺序运行,以及如何移交工作,并且在这些方面做得非常深入。SwarmMesh 不做这些;它只为已在运行的 Agent 提供一个共享位置,用于读写状态。feynman 是一款单一的研究 Agent 产品,带有本地工作台 UI 和自己的 SQLite 支持状态,并非供其他项目嵌入的协调层。与 SwarmMesh 的 docs/protocol.md 一样,这两个项目都没有提供有文档说明的跨语言线协议,用于共享 Agent 记忆。完整的并排对比请见上文的 SwarmMesh 对比。

记忆搜索是语义搜索/基于嵌入向量的吗? 不是。它使用 Okapi BM25 关键词排序算法,与搜索引擎几十年来使用的算法属于同一类,在本地根据词频计算。对于与你的查询概念相关、但没有共享词汇的记忆条目,它无法检索到。如果你需要这种能力,RankingBackend 接口是一个有文档说明的扩展点,可用于接入你自己的基于嵌入向量的评分器。SwarmMesh 不自带此类评分器,也不会在未告知你的情况下代你调用嵌入 API。

Python Agent 和 Node Agent 真的能共享状态吗?还是说这只是理论上的? 这正是这个项目存在的原因。两个 CLI 都实现了 docs/protocol.md 中定义的同一套线协议,而且上文的“实际运行效果”部分提供了一份真实记录:Node CLI 将上下文和记忆写入由 Python 托管的服务,然后 Python CLI 通过网络读回这些内容;编写本 README 时已再次验证。

SwarmMesh 会持久化数据吗? 只有在你要求时才会。swarmmesh serve 默认使用内存存储,进程退出后数据就会消失。传入 --persist <path> 可使用基于 SQLite 的存储,数据会在重启后保留。

有身份验证吗? v1 中没有。请参阅下文的安全性:这是有意设定的范围边界,并非疏忽。

如果两个 Agent 向同一个上下文键写入数据,会发生什么? 后写入的内容会覆盖先写入的内容。PUT /v1/context/{namespace}/{key} 会覆盖原有内容。每次写入都会广播一个 context.updated WebSocket 事件,因此订阅该命名空间的 Agent 会立即获知,而不必轮询。系统没有合并或冲突解决机制;如果你的 Agent 需要这些功能,可使用不同的键或自己的版本管理约定来实现。

我可以把 SwarmMesh 作为库使用,而不是通过 CLI 使用吗? 可以。两个软件包都导出了客户端:Python 中的 swarmmesh_cli.client.SwarmMeshClient,以及 Node 中来自 swarmmesh-cli 的 SwarmMeshClient。如需查看真实的方法签名,请参阅上文的库 API 参考。

我可以在多台机器上运行它吗?它已经可以用于生产环境了吗? 没有什么会阻止网格通过网络访问;--host 可绑定到你指定的任意网络接口。但 v1 没有身份验证(见安全性),因此应将它视为本地 Redis 实例,而不是面向公共互联网的服务。目前已知的生产用户数量也是 0,请据此评估。

可以免费用于商业用途吗? 可以。Python 和 Node 软件包以及仓库本身均采用 MIT 许可。无需申请许可或支付费用,即可将其用于商业产品。

安全性

警告

SwarmMesh v1 不提供身份验证。未经反向代理添加身份验证,便将 SwarmMesh 服务直接暴露到公共互联网,是一种错误配置,并非受支持的部署方式。

Python 和 Node 服务默认都绑定到 127.0.0.1,而非 0.0.0.0。SwarmMesh 旨在 localhost 上运行,或与其协调的 Agent 一同在私有 网络内部运行。这与本地 Redis 实例或 SQLite 文件处于相同的信任边界, 而不是面向公共互联网的服务。

发现漏洞?请勿提交公开 issue。有关私下披露流程,请参阅 SECURITY.md。

贡献

SwarmMesh 有两个官方实现,遵循相同的协议,并刻意保持行为一致。有关两者的 开发环境设置、拉取请求流程,以及塑造本 README 的基本原则——不得提出未经 验证的声明——请参阅 CONTRIBUTING.md。此处的每个 数字都必须能通过实际命令复现。

许可证

MIT

评论

支持 Markdown

第一次接触 MCP?

从协议是什么,到在客户端里装上第一个服务,这几条回答覆盖了大多数入门问题。

一个开放协议,规定了 AI 应用如何以统一方式连接外部工具与数据源。客户端负责与模型对话,MCP 服务负责提供工具、资源和提示模板,两者通过标准消息通信,同一个服务可以被不同客户端复用。