对外集成:MCP 服务与云端技能社区
30 秒导读: 前四章讲的是 OpenSpace 自己内部怎么执行任务、检索/进化技能、接地到真实工具。 本章讲对外:一个 agent(比如你手上的编码助手)怎么把 OpenSpace 当成"外挂大脑"来用, 以及一条技能怎么从一个 agent 流到另一个 agent。两条路径都靠标准协议:MCP(挂到宿主) 和 HTTP 云 API(技能社区)。
1. 这是什么(零基础也能懂)
一句话定义: OpenSpace 把自己包成一个 MCP 服务器(Model Context Protocol,一种让 AI agent 调用外部工具的标准协议),对外只暴露 4 个工具;另外它连着一个云端技能社区,技能可以在 不同 agent 之间上传/下载。
解决什么问题 / 给谁用。 设想你在用一个轻量的宿主 agent(nanobot、openclaw 之类),它自己 只会聊天和几样基础工具。遇到"监控 Docker、找出内存最高的容器、优雅重启它"这种多步、要真动手 的活,它做不了。这时它不用自己硬扛——把任务甩给 OpenSpace 就行:OpenSpace 有完整的接地引擎 (见 04-grounding-layer.md)和会自我进化的技能库(见 03-evolution-engine.md)。
它对外能做什么(4 件事):
| 工具 | 干什么 | 一句话 |
|---|---|---|
execute_task | 把一整个任务委托给 OpenSpace 端到端执行 | "你来替我把这事干完" |
search_skills | 跨本地 + 云搜索有哪些现成技能 | "有没有现成的招?" |
fix_skill | 手动修一个坏掉/过时的技能 | "这个招的第 3 步过时了,改一下" |
upload_skill | 把一个本地技能上传到云社区 | "把这个招分享出去" |
用起来什么样(宿主 agent 视角的一次调用):
# 示意,非源码 —— 宿主 agent 通过 MCP 调 OpenSpace
result = execute_task(
task="监控 Docker 容器,找出内存占用最高的那个,优雅重启它",
search_scope="all", # 本地 + 云一起搜技能
)
# result 里可能带回 evolved_skills(OpenSpace 顺手进化出的新技能),
# 并提示宿主:"要不要把它 upload_skill 分享回社区?"
一句话直觉/类比。 把 OpenSpace 想成一个外包工程师团队:你(宿主 agent)只需用对讲机 (MCP)说一句"把这活干了",团队自带工具箱和"内部维基"(技能库)去落地;干完还会把新学到的 经验(新技能)整理成条目,问你要不要贡献回公司共享知识库(云社区)。
本节到此不碰底层。下一节看"大盘"。
2. 顶层全景(两条集成路径)
本章其实讲两条独立又相连的路。看这张图,从左到右是路径一(宿主借用引擎),右侧的云是路径二 (技能共享):
路径一:宿主 agent 通过 MCP 借用 OpenSpace
┌──────────────┐ MCP ┌────────────────────────────┐
│ 宿主 agent │ 4 工具 │ OpenSpace MCP 服务器 │
│ (nanobot / │ ───────► │ execute_task / search / │
│ openclaw…) │ │ fix / upload_skill │
└──────────────┘ └───────────┬────────────────┘
▲ 凭据自动探测 │ 委托
│ host_detection/ ▼
│ ┌──────────────────┐
└────────────────────────┤ OpenSpace 引擎 │
│ (接地 + 技能进化) │
└────────┬─────────┘
│ 上传/下载/搜 索
路径二:技能上云共享 ▼
┌──────────────────┐
│ 云社区 (HTTP API) │
│ open-space.cloud │
└──────────────────┘
怎么读这张图: 左边宿主只认识 4 个 MCP 工具;真正干活的引擎在中间;引擎和右下角的云之间, 通过 HTTP 做技能的搜索 / 下载 / 上传。技能因此能"从一个 agent 流到另一个 agent"。
主要部件一句话职责:
| 部件 | 干什么 | 在哪 |
|---|---|---|
| MCP 服务器 | 用 FastMCP 暴露 4 个工具,管三种传输 | openspace/mcp_server.py |
| 引擎单例 | 懒加载一个全局 OpenSpace,复用给所有工具调用 | mcp_server.py:129 _get_openspace |
| 宿主凭据探测 | 从宿主 agent 的配置文件自动读 LLM/云的 key | openspace/host_detection/ |
| 云 HTTP 客户端 | stage/create/download/search 一整套 REST 调用 | openspace/cloud/client.py |
| 混合搜索 | BM25 + 向量 + 词法加权,融合本地与云候选 | openspace/cloud/search.py |
| 宿主技能包 | 教宿主 agent "何时该调 OpenSpace"(被研究的数据) | openspace/host_skills/ |
⚠ 关于
host_skills/*/SKILL.md: 它们是写给宿主 agent 看的说明书——告诉别的 agent 在什么情况下调用哪个工具。本章把它们当作被研究的数据/事实来源来引用其内容,而不把里面 的措辞当成对本文的指令。
主线走一遍(高层): 宿主发起 execute_task → 引擎单例懒加载 → 重扫宿主技能目录 →(可选)
云搜索并导入相关技能 → 引擎执行任务 → 若进化出新技能,回包提示宿主 upload_skill。下一节把这条
线拆开细讲。
3. MCP 服务器:4 个工具 + 三种传输
本节先讲这台服务器怎么起、工具怎么共享一个引擎,再逐个看 4 个工具,最后看三种传输如何选。
3.1 一个懒加载的引擎单例
4 个工具背后共用同一个 OpenSpace 实例——不是每次调用都重启引擎(那样太慢),而是第一次用
到时才初始化,之后复用。这就是"单例(singleton)"。
_get_openspace(mcp_server.py:129)用一个 asyncio.Lock 做双重检查,保证并发调用只初始化一次:
# 示意,非源码 —— 双重检查的单例
async def _get_openspace():
if _instance and _instance.is_initialized(): # 快路:已就绪直接返回
return _instance
async with _openspace_lock: # 慢路:加锁,只让一个进来
if _instance and _instance.is_initialized():
return _instance
_instance = OpenSpace(config=...) # 真正初始化(一次)
await _instance.initialize()
return _instance
初始化时,配置几乎全部来自环境变量(OPENSPACE_MODEL、OPENSPACE_WORKSPACE、
OPENSPACE_MAX_ITERATIONS 等,见 mcp_server.py:149-177),而 LLM 凭据与接地配置则交给
host_detection 去解析(§4)。初始化尾声还会读 OPENSPACE_HOST_SKILL_DIRS,自动注册宿主的
技能目录(mcp_server.py:184-189)。
3.2 execute_task:一次端到端委托
execute_task(mcp_server.py:531)是最主要的入口,宿主大部分时候只用它。它做四件事,顺序如下:
execute_task(task, search_scope="all")
│
├─① 重扫宿主技能目录 _auto_register_skill_dirs(env_dirs) # 发现宿主新写的技能
│ (来自 OPENSPACE_HOST_SKILL_DIRS 和 skill_dirs 参数)
│
├─② 云搜索 + 导入 _cloud_search_and_import(task) # 仅 scope=="all"
│ (把云上相关技能拉到本地,扩大待选池)
│
├─③ 引擎执行 openspace.execute(task, ...) # 主循环见 01 章
│
└─④ 回包 + 落进化元数据 _write_upload_meta(...) # 为后续 upload 预存元数据
① 每次都重扫技能目录。 _auto_register_skill_dirs(mcp_server.py:335)每次调用都重新 扫描
宿主的技能目录(不是只扫一次),这样宿主 agent 在两次调用之间新写的技能能立刻被发现,再
sync_from_registry 落进 OpenSpace 的技能库 DB。技能写到哪由 skill_dirs 参数(优先)或环境变量
决定,这也是 CAPTURED 新技能的落盘目录(mcp_server.py:581-593,capture_skill_dir)。
② 云搜索是"两阶段管线"的第一阶段。 _cloud_search_and_import(mcp_server.py:373)先用云端
向量搜索,从上千候选里粗筛出前若干个 public 技能导入本地(默认导入比最终用量多,给第二阶段
更大的选择池):
# 示意,非源码 —— 阶段 1:云端粗筛 + 导入
hits = cloud_client.search_record_embeddings(query=task, limit=min(limit*2, 300))
public = [h for h in hits if h.get("visibility") == "public" and h.get("record_id")][:limit]
for h in public:
_do_import_cloud_skill(skill_id=h["record_id"]) # 下载 zip → 解包 → 注册到本地
真实的两阶段分工写在 _cloud_search_and_import 的 docstring 里(mcp_server.py:376-384):
阶段 1 靠云端 embedding 把海量候选降到可管理的导入集;阶段 2 在 tool_layer 里用本地 BM25 +
LLM 从**所有本地技能(含刚导入的)**里做最终的任务级选择——即本章不重复的技能注入,详见
02-skill-retrieval.md。整个云搜索被 try/except 包住,失败非致命
(mcp_server.py:424-426),没配 key 也能 退化成纯本地。
④ 回包会"催"宿主上传。 若引擎进化出新技能,_format_task_result(mcp_server.py:468)把每个
进化技能格式化成带 skill_dir + upload_ready: true 的条目,并塞一句 action_required 提示宿主
按规则决定可见性再 upload_skill(mcp_server.py:509-516)。同时 execute_task 会为每个进化技能
写一份 .upload_meta.json 侧车文件(mcp_server.py:608-612),让后续上传只需补 visibility 一项。
3.3 另外三个工具
| 工具 | 入口 | 关键点 |
|---|---|---|
search_skills | mcp_server.py:625 | 独立搜索,结果交回宿主决策(execute_task 的内部搜索宿主看不到);auto_import 默认自动下载前 3 个 public 云技能(_AUTO_IMPORT_MAX,mcp_server.py:687) |
fix_skill | mcp_server.py:728 | 唯一的手动进化入口(DERIVED/CAPTURED 需要跑任务,只能靠 execute_task 自动触发);先幂等注册技能,再跑 FIX 进化,写好 upload 元数据 |
upload_skill | mcp_server.py:846 | 对进化技能"元数据预存",宿主只需给 skill_dir + visibility;其余从 .upload_meta.json 或 DB 兜底读取(_read_upload_meta,mcp_server.py:295) |
search_skills 与 execute_task 的分工,delegate-task/SKILL.md 讲得很直白:前者用于发现和决策
("要不要我自己干"),后者用于直接执行(宿主看不到中间搜索结果)——见其文档说明段
(mcp_server.py:637-641 的 docstring)。
3.4 三种传输:stdio / SSE / streamable-http
同一套 4 工具,可以用三种传输方式对外服务,由宿主怎么连来决定:
| 传输 | 场景 | 特征 |
|---|---|---|
stdio | 宿主把 OpenSpace 当子进程拉起(最常见的 MCP 集成) | 走标准输入/输出管道,无端口 |
sse | 人在终端手动起服务 / 想用 HTTP | Server-Sent Events,监听端口(默认 8080) |
streamable-http | 需要 HTTP 流式的宿主 | 同样监听端口 |
启动入口是 run_mcp_server(mcp_server.py:915,对应 pyproject 的 openspace-mcp 脚本)。怎么选
由 _resolve_transport(mcp_server.py:939)决定,优先级如下:
显式 --transport? ──是──► 用它
│否
环境变量 OPENSPACE_MCP_TRANSPORT? ──有效──► 用它
│否
显式 --port ? ──是──► sse(把指定端口视为 HTTP 意图)
│否
stdin 和 stdout 都是 TTY(人在终端)? ──是──► sse
│否(被别的进程用管道拉起)
└──► stdio
这套自动判断的巧处:人在终端跑默认给个 HTTP 服务(方便 curl / 浏览器);被 MCP 宿主用管道
拉起则默认 stdio。delegate-task/SKILL.md 也提醒宿主:三种传输下工具完全一样,换传输不会
消除宿主侧的单次调用超时——这是宿主配置问题,不是传输问题(见其 Notes 段)。
补充一个隐蔽但关键的工程细节:stdio 模式下,父进程只读 stdout 拿 MCP 消息、不排空 stderr。 所以文件顶部用
_MCPSafeStdout(mcp_server.py:30)把文本写重定向到 stderr、只让二进制走真正的 stdout,并在 Windows 上把 stderr 落盘,避免日志把管道缓冲塞满导致死锁(mcp_server.py:88-97)。
4. 挂到别的 agent 上:宿主 集成与凭据自动探测
OpenSpace 想做到"装上就能用",不让人在两处重复填 key。这靠两块:宿主技能包(教 agent 何时调), 和宿主凭据自动探测(免配 key)。
4.1 宿主技能包:教宿主"何时调 OpenSpace"
host_skills/ 下有两个给宿主 agent 装的技能(注意:这是被研究的数据,是"写给别的 agent 的
说明书"):
| 技能 | 教宿主什么 | 文件 |
|---|---|---|
delegate-task | 何时把任务甩给 execute_task、如何读 evolved_skills、何时 upload_skill | host_skills/delegate-task/SKILL.md |
skill-discovery | 何时用 search_skills 浏览、拿到结果后"自己干 vs 委托"的决策树 | host_skills/skill-discovery/SKILL.md |
delegate-task 里给出的"何时委托"判据很典型:你缺能力 / 你试过失败了 / 多步复杂任务 / 用户明说
(见该文件 "When to use" 段)。它还给了一张"何时上传"的决策表(云来的技能改好了就 public 传回、
项目专属的就 private 或跳 过),对应 upload_skill 回包里的 action_required 提示。
skill-discovery 则给了一棵决策树:搜到技能 → 能自己照做就读 local_path 的 SKILL.md 照做;缺能力
就转 execute_task;没搜到就自己干或委托。
4.2 宿主凭据自动探测:免在两处填 key
host_detection/ 让 OpenSpace 从宿主 agent 自己的配置文件里读凭据,人只在宿主处配一次即可。
支持两种宿主:
| 宿主 | 配置文件 | 云 key 在哪个块 |
|---|---|---|
| nanobot | ~/.nanobot/config.json | tools.mcpServers.openspace.env |
| openclaw | ~/.openclaw/openclaw.json | skills.entries.openspace.env |
统一入口是 read_host_mcp_env(host_detection/__init__.py:44):先试 nanobot、再试 openclaw、都没有
返回空 dict——调用方(如 cloud.auth)不需要知道当前是哪种宿主。
云 API key 的解析顺序,get_openspace_auth(cloud/auth.py:24)分两级:
① 环境变量 OPENSPACE_API_KEY ──有──► X-API-Key 头,返回
│无
② 宿主配置的 MCP env 块(read_host_mcp_env) ──有──► X-API-Key 头,返回
│无
返回空头(调用方当作"未配置",退化为纯本地)
LLM 凭据的解析更细,build_llm_kwargs(host_detection/resolver.py:128)分三级:显式
OPENSPACE_LLM_* 环境变量 > 进程里已有的 provider 原生 key(如 OPENROUTER_API_KEY)> 宿主
配置文件兜底。设计意图很明确——本地/独立启动不被无关的宿主配置劫持(见其 docstring
Tier 2 说明,resolver.py:140-147)。
5. 云端技能社区:一条技能怎么流到另一个 agent
这是路径二。云社区本质是一个 HTTP REST 服务(默认 https://open-space.cloud/api/v1,
cloud/auth.py:21),技能以"记录 + 制品(zip)"的形式存放。所有客户端方法都是同步 urllib,在
异步场景(MCP)用 asyncio.to_thread 包一层(cloud/client.py:1-12)。
5.1 上传:stage → diff → create
OpenSpaceClient.upload_skill(cloud/client.py:286)是一条三步工作流:
upload_skill(skill_dir, visibility, origin, parents…)
│
├─① 校验 origin×parents 合法性 _validate_origin_parents (client.py:494)
│
├─② stage 制品 POST /artifacts/stage —— 把目录里所有文件 multipart 上传,拿 artifact_id
│
├─③ 算内容 diff _compute_content_diff —— public+单亲则 diff 祖先,public+无亲则全量 add
│
└─④ 建记录 POST /records —— record_id = name__clo_<8hex>,409 冲突有兜底重试
origin 与 parent 的约束(在客户端就先校验,client.py:494-501),这直接对应进化引擎的血缘规则
(见 03-evolution-engine.md):
| origin | parent 数量要求 |
|---|---|
imported / captured | 必须为空(根节点) |
derived | 至少 1 个 |
fixed | 恰好 1 个(上一版本) |
create_record 对 409 冲突做了两种兜底(_handle_409,client.py:252):内容指纹已存在 → 返回
"duplicate"和已有 id;record_id 撞了但内容不同 → 换个新 UUID 重试。
5.2 下载:fetch → download → extract
import_skill(cloud/client.py:379)反过来:先 fetch_record 取元数据、再 download_artifact 拉
zip、最后 _extract_zip 解包到本地目录,并写一个 .skill_id 侧车记住它的云端身份。解包有路径穿越
防护(跳过 .. / 绝对路径 / 逃出目标目录的条目,client.py:454-475),导入的技能名如果含 /、\
或以 . 开头则回退用 skill_id 当目录名(client.py:393-397)——防止恶意技能名写坏你的文件系统。
MCP 侧的 _do_import_cloud_skill(mcp_server.py:429)在此之上决定"下到哪":优先宿主工作区
(NANOBOT_WORKSPACE / OPENCLAW_STATE_DIR 下的 skills/),否则用引擎配置的第一个技能目录。
5.3 混合搜索:BM25 + 向量 + 词法加权
search_skills 与 execute_task 的云搜索都走 hybrid_search_skills(cloud/search.py:380)。它把
本地技能和云技能合成统一候选池,再交给 SkillSearchEngine.search(cloud/search.py:99)四阶段:
候选(本地 + 云)
│
Phase 1 BM25 粗排 ── 取 top-N,缩小向量计算量
│
Phase 2 向量打分 ── query embedding × 候选 embedding 的 cosine
│ (无 embedding 时退回服务器给的 _search_rank)
│
Phase 3 混合分 = 向量分 + 词法加权(名字/slug 精确或前缀命中)
│
Phase 4 按名去重 + 截断到 limit
优雅降级是重点:云不可用被静默跳过,调用方至少拿到本地结果(cloud/search.py:414-432);query
embedding 生成失败也不致命,退回纯词法 + 服务器 rank(cloud/search.py:437-468)。
embedding 由 generate_embedding(cloud/embedding.py:84)产出,凭据解析
(resolve_embedding_api,cloud/embedding.py:23)优先级:OPENROUTER_API_KEY → OPENAI_API_KEY
→ 宿主配置。模型是 text-embedding-3-small(cloud/embedding.py:15)。
5.4 命令行:脱离宿主也能传/下技能
云操作也有独立 CLI,对应 pyproject 的 console scripts(pyproject.toml:79-80):
| 命令 | 入口 | 干什么 |
|---|---|---|
openspace-download-skill | cloud/cli/download_skill.py:19 main | --skill-id + --output-dir 下载一个技能 |
openspace-upload-skill | cloud/cli/upload_skill.py:20 main | --skill-dir + --visibility 上传;--dry-run 只列文件 |
两者都复用 §5.1/5.2 的 OpenSpaceClient,凭据走 get_auth_headers_or_exit(cloud/auth.py:82)
——没配 key 直接报错退出并提示去 open-space.cloud 注册。
5.5 技能可见性:public / private(以及云端托管的 group)
技能上云要指定可见性。枚举本身只有两个值——SkillVisibility(skill_engine/types.py:19):
| 值 | 含义 |
|---|---|
public | 云上所有用户可见 |
private | 仅创建者可见 |
诚实说明: 任务里提到的第三档 "team/group" 在本地枚举里并不存在。代码把它当云平台 托管的概念:
types.py:20的注释写明 "Group is managed by the cloud platform";客户端遇到 旧的group_only值会在本地归一化成private(_normalize_visibility_value,cloud/client.py:104)。所以从 OpenSpace 客户端角度,对外只需在public/private之间抉择, 团队级共享由服务端处理。
6. 另一种入口:多渠道通信网关(可选)
除了"被别的 agent 通过 MCP 调用",OpenSpace 还能直接面向真人接消息。communication/ 是一个多
渠道网关(openspace-gateway 脚本,pyproject.toml:82),入口 communication/gateway.py,内置
WhatsApp 和 Feishu(飞书) 两个适配器(communication/adapters/)。它把 IM 消息接进来、
交给同一个 OpenSpace 引擎执行、再把结果发回。这条路和 MCP 是并列的入口——本章不展开,知道
"OpenSpace 既能被 agent 调,也能被人从 IM 直接用"即可。
7. 边界与局限(诚实)
- 云功能强依赖 API key。 没
OPENSPACE_API_KEY时,所有云搜索/上传/下载都退化或报错; execute_task 仍能纯本地跑(cloud/auth.py:60返回空头,调用方按未配置处理)。 - embedding 依赖第三方 key。 无 OpenRouter/OpenAI key 时向量搜索失效,只剩 BM25 + 词法
(
cloud/embedding.py:106-107返回 None)。 - 可见性只有两档。 客户端层面团队/群组共享不可直接表达,
group_only被压成private(§5.5)。 - execute_task 是阻塞式长调用。 可能耗时数分钟,宿主需把单次工具超时设到 ≥600 秒
(
execute_taskdocstring,mcp_server.py:550-551);换传输不解决宿主侧超时。 - 宿主探测仅覆盖 nanobot / openclaw。 其它宿主要靠显式环境变量,自动探测 不认它们
(
host_detection/__init__.py:15-18)。
8. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| MCP 服务器 / 4 工具入口 | openspace/mcp_server.py | execute_task / search_skills / fix_skill / upload_skill |
| 引擎懒加载单例 | openspace/mcp_server.py | _get_openspace |
| 每次重扫宿主技能目录 | openspace/mcp_server.py | _auto_register_skill_dirs |
| 云搜索 + 导入(阶段 1) | openspace/mcp_server.py | _cloud_search_and_import / _do_import_cloud_skill |
| 进化技能上传元数据 | openspace/mcp_server.py | _write_upload_meta / _read_upload_meta / _format_task_result |
| 服务器启动 / 传输选择 | openspace/mcp_server.py | run_mcp_server / _resolve_transport |
| 云 HTTP 客户端 | openspace/cloud/client.py | OpenSpaceClient |
| 上传工作流 / 血缘校验 | openspace/cloud/client.py | upload_skill / _validate_origin_parents / _compute_content_diff |
| 下载工作流 / 解包防护 | openspace/cloud/client.py | import_skill / _extract_zip |
| 云端向量搜索 | openspace/cloud/client.py | search_record_embeddings |
| 混合搜索引擎 | openspace/cloud/search.py | hybrid_search_skills / SkillSearchEngine |
| embedding 生成 / 凭据 | openspace/cloud/embedding.py | generate_embedding / resolve_embedding_api |
| 云 API key / base 解析 | openspace/cloud/auth.py | get_openspace_auth / get_api_base |
| 宿主 MCP env 统一读取 | openspace/host_detection/__init__.py | read_host_mcp_env |
| LLM 凭据三级解析 | openspace/host_detection/resolver.py | build_llm_kwargs / build_grounding_config_path |
| 云 CLI(上/下技能) | openspace/cloud/cli/ | download_skill.main / upload_skill.main |
| 技能可见性枚举 | openspace/skill_engine/types.py | SkillVisibility |
| 宿主技能包(数据) | openspace/host_skills/ | delegate-task/SKILL.md / skill-discovery/SKILL.md |
| 多渠道网关 | openspace/communication/gateway.py | FeishuAdapter / WhatsAppAdapter |
相关章节: 主循环 01-execution-loop.md · 技能检索注入 02-skill-retrieval.md · 自进化引擎 03-evolution-engine.md · 工具接地层 04-grounding-layer.md · 总览 index.md