进阶 omnigent.ai 2026-10-14 06:47:13 · 17 阅读

第28章编程式使用 Omnigent

编程式使用

将 Omnigent 集成到你的脚本、应用程序和 CI 流水线中。你可以启动 Agent 会话，向其分派任务，并返回同一会话进行后续操作。此外，你还可以在 Web UI 中打开已保存的会话，查看其进度和结果。

CLI：在命令行运行任务并接收 Agent 响应。

Python SDK：从 Python 发送消息和管理会话。

REST API：通过 HTTP 创建会话、读取历史及管理生命周期。

开始之前：安装 Omnigent 并配置 Agent 所需的模型凭证。示例使用运行在 http://127.0.0.1:6767 的本地单用户服务器。若使用共享服务器，请替换该 URL，并在发送请求前配置身份验证。详见“共享服务器”部署指南。本地单用户服务器无需为每个请求提供凭证。安装完整的 omnigent Python 包或仅安装 omnigent-client 即可使用 SDK。两者均提供下文使用的 omnigent_client 导入功能。请使用与服务器版本匹配的包版本。

共享服务器的身份验证：共享服务器支持用户认证和可选的机器令牌。所使用的凭证取决于服务器的身份验证配置：

账户或 SSO：使用已登录用户的会话 Cookie 或 CLI 登录 Bearer 令牌进行 API 请求。这些凭证支持定时任务及会话，权限受限于用户配置。
omni login ""
此命令提示输入内置账户的用户名和密码，或针对 SSO 打开浏览器。它保存凭证供后续 CLI 命令使用，不会输出用于 curl 的令牌。
对于内置账户，你也可以直接请求令牌：
ACCESS_TOKEN=$(curl --fail-with-body "/auth/login" \
   -H 'Content-Type: application/json' \
   -d '{"username": "", "password": ""}' \
   | jq -er '.token')
在 API 请求中使用 -H "Authorization: Bearer $ACCESS_TOKEN"。此用户名/密码端点不适用于 SSO。
身份验证代理：通过代理连接，由其向 Omnigent 提供身份信息。遵循代理提供商的身份验证方法；Omnigent 在此模式下不颁发令牌。
机器令牌：对于无人值守脚本，管理员可启用机器访问并提供客户端 ID 和密钥。这些令牌支持 agent、host、runner 和 session API，但无法访问定时任务。
curl --fail-with-body "/oauth/token" \
   --data-urlencode grant_type=client_credentials \
   --data-urlencode "client_id=" \
   --data-urlencode "client_secret="
在 API 请求或 Python SDK 头中将返回的 access_token 作为 Authorization: Bearer 包含。令牌默认一小时过期；运行相同命令可获取新令牌。

启动会话：从仓库中运行 Agent，使用 -p 发送提示词，打印响应并退出。使用 Agent 目录或 YAML 文件替代 ./my-agent/：
omni run ./my-agent/ -p "Review the changes in the latest commit"
Omnigent 会自动启动本地服务器和 Runner。会话默认保存；启动时其 URL 显示在 stderr 上。打开它可在 Web UI 中跟踪进度：
Omnigent session: http://127.0.0.1:6767/c/conv_abc123

常用选项：

选项	适用场景
`-p ""` / `--prompt ""`	非交互式运行任务；响应输出到 stdout。
`--server ""`	在共享服务器上保存会话。Agent 定义会被上传，工具仍在调用 CLI 的机器上运行。
`--no-session`	运行一次性本地任务，事后无历史记录可查或恢复。
`--model ""`	覆盖 Agent 的模型。参见“模型与凭证”。
`--harness ""`	覆盖 Agent 的 Harness。参见“Harnesses”了解支持的配置。
`--continue / -c`	继续同一 Agent 最近的会话。
`--resume ""`	使用会话 URL 中的 ID 继续特定会话。

这些示例通过 omni run 使用 Agent YAML 或目录；原生终端 Harness 有其自身的 CLI 行为。若要在其他机器或云端沙箱中运行工具，请使用 REST API 的 host 选项。

此示例在现有服务器上创建原生 Claude Code 会话。你需要一个已配置好仓库、凭证和工具的已连接 Runner。


import asyncio
from omnigent_client import OmnigentClient
 
async def main():
    async with OmnigentClient(
        base_url="http://127.0.0.1:6767",
        # 对于需要 Bearer 令牌的服务器：
        # headers={"Authorization": "Bearer "},
    ) as client:
        agent = await client.sessions.resolve_agent("claude-native-ui")
        runner_id = await client.sessions.resolve_online_runner(harness=agent.harness)
        if runner_id is None:
            raise RuntimeError("Connect a runner that supports native Claude Code first")
 
        session = await client.sessions.create_from_agent_id(agent.id, title="Code review")
        await client.sessions.bind_runner(session.id, runner_id=runner_id)
        print(session.id)
        print(f"http://127.0.0.1:6767/c/{session.id}")
 
asyncio.run(main())

使用打印出的会话 ID 在下一步发送消息，或打开 URL 查看会话。

使用 curl 和 jq 创建原生 Claude Code 会话。连接包含仓库的机器，并在使用该会话时保持此命令运行：
omni host --server http://127.0.0.1:6767
对于共享服务器，请在每个请求中包含身份验证凭证，例如 -H "Authorization: Bearer $ACCESS_TOKEN"。

使用场景	示例
查找 Agent	获取原生 Claude Code 的 Agent ID： `curl -fsS http://127.0.0.1:6767/v1/agents \` `\| jq -r '.data[] \| select(.harness == "claude-native") \| .id'`
查找已连接的主机	列出主机名称和 ID： `curl -fsS http://127.0.0.1:6767/v1/hosts \` `\| jq '.hosts[] \| {name, host_id}'` 获取特定主机的 ID： `curl -fsS http://127.0.0.1:6767/v1/hosts \` `\| jq -r '.hosts[] \| select(.name == "") \| .host_id'`
在已连接的主机上启动	使用 Agent ID、主机 ID 和仓库路径。`SESSION_ID` 存储新会话 ID： `SESSION_ID=$(curl --fail-with-body http://127.0.0.1:6767/v1/sessions \` `-H 'Content-Type: application/json' \` `-d '{` `"agent_id": "ag_reviewer",` `"host_id": "host_example",` `"workspace": "/home/ci/my-repository",` `"title": "Review latest commit"` `}' \| jq -er '.id')`
在云沙箱中启动	对于已配置的云沙箱主机，在上述创建请求中使用 `host_type: "managed"` 替代 `host_id`。服务器会异步配置主机。
检查会话是否就绪	在发送消息前检查 `runner_online` 是否为 true： `curl --fail-with-body "http://127.0.0.1:6767/v1/sessions/$SESSION_ID" \| jq`

发送消息或继续会话：使用上一步的会话 ID 向该会话发送消息。已保存的会话也可在 Web UI 中打开。使用 --continue (或 -c) 向同一 Agent 最近的会话发送后续提示词：
omni run ./my-agent/ --continue -p "Now check whether the tests cover those changes"
要继续特定会话，请使用其 ID：
omni run ./my-agent/ --resume "" -p "Explain the first finding"
当自动化运行多个会话且需指定目标时，使用 --resume。使用 CLI 会话 URL 中的 ID，或通过 GET /v1/sessions 查找。

此示例向现有会话发送消息。将 conv_abc123 替换为其 ID。会话必须有一个拥有仓库、模型凭证和工具访问权限的在线 Runner。SDK 不会为你启动 Runner。


import asyncio
from omnigent_client import OmnigentClient
 
async def main():
    async with OmnigentClient(base_url="http://127.0.0.1:6767") as client:
        await client.sessions.post_event("conv_abc123", {
            "type": "message",
            "data": {
                "role": "user",
                "content": [{"type": "input_text", "text": "tell me a joke"}],
            },
        })
 
asyncio.run(main())

该调用在服务器接受消息时返回，不会等待 Agent 回复。在 Web UI 中打开会话以跟踪其进度。

向同一会话发送初始提示词或后续消息：
curl --fail-with-body "http://127.0.0.1:6767/v1/sessions/$SESSION_ID/events" \
   -H 'Content-Type: application/json' \
   -d '{
     "type": "message",
     "data": {
       "role": "user",
       "content": [{"type": "input_text", "text": "Review the changes in the latest commit"}]
     }
   }'
请求在服务器接受消息时返回，不等待回复。打印 URL 以跟踪审查进度：
echo "http://127.0.0.1:6767/c/$SESSION_ID"

查找会话并跟踪进度：CLI 启动时将会话 URL 打印到 stderr，运行结束时将收集的助手文本打印到 stdout。打开该 URL 可对话及进度。使用 Python 或 REST 选项卡列出已保存会话、获取历史或从应用程序监控 Agent。

列出最近会话，然后读取会话状态和消息。将 conv_abc123 替换为要检查的 ID。


import asyncio
from omnigent_client import OmnigentClient
 
async def main():
    async with OmnigentClient(base_url="http://127.0.0.1:6767") as client:
        sessions = await client.sessions.list(limit=20, sort_by="updated_at")
        for session in sessions:
            print(session.id, session.title, session.status)
        session_id = "conv_abc123"
        session = await client.sessions.get(session_id)
        print(session.status)
        print(await client.sessions.list_items(session_id))
 
asyncio.run(main())

对于实时事件，迭代 client.sessions.stream(session_id)。使用 session.status 检查会话本身，使用 await client.sessions.subtree_busy(session_id) 检查其任何子节点是否忙碌。历史是分页的。使用最后一项的 id 作为下一次 list_items 调用中的 after 参数以读取更多。实时事件不重放历史；重新连接后需重新获取快照和历史。在本地单用户服务器上列出会话：


curl --fail-with-body "http://127.0.0.1:6767/v1/sessions?limit=20&sort_by=updated_at"

可按 agent_name、agent_id 或 search_query 过滤。列表默认排除已归档会话和子会话；如需包含，使用 include_archived=true 或 kind=any。会话和历史列表均分页：当 has_more 为 true 时，使用返回的 last_id 作为下一次请求的 after 游标。读取会话历史：


curl --fail-with-body "http://127.0.0.1:6767/v1/sessions//items"

SSE 流提供实时更新；不重放对话历史。重新连接后使用会话快照和 /items 恢复状态。

分叉、归档或停止工作：分叉为交互式会话：
omni run ./my-agent/ --fork ""
--fork 不能与 -p 组合使用。对于脚本化分叉、归档或取消，请使用 Python 或 REST 选项卡。

这些调用在上述 async with OmnigentClient(...) as client 块中运行。将 session_id 替换为要管理的会话。

操作	SDK 调用
取消活动工作	`await client.sessions.interrupt(session_id)`
分叉会话	`fork = await client.sessions.fork(session_id)`
归档并保留历史	`await client.sessions.set_archived(session_id, archived=True)`
恢复已归档会话	`await client.sessions.set_archived(session_id, archived=False)`

分叉的新 ID 为 fork["id"]。发送提示词前需绑定 Runner：await client.sessions.bind_runner(fork["id"], runner_id=runner_id)。分叉默认复制完整历史。传入 up_to_response_id 可从更早的响应分叉。使用 REST 选项卡可永久删除会话及其资源。

使用早期示例中的 SESSION_ID。

使用场景	示例
取消活动工作	停止当前任务而不删除会话： `curl --fail-with-body "http://127.0.0.1:6767/v1/sessions/$SESSION_ID/events" \` `-H 'Content-Type: application/json' \` `-d '{"type": "interrupt"}'`
分叉会话	复制会话并将新会话 ID 存入 `FORK_ID`： `FORK_ID=$(curl --fail-with-body "http://127.0.0.1:6767/v1/sessions/$SESSION_ID/fork" \` `-H 'Content-Type: application/json' \` `-d '{"title": "Explore another approach"}' \| jq -er '.id')` 这会复制完整历史。在请求体中添加 `"up_to_response_id": ""` 以仅复制至该响应。发送另一个提示词前需为分叉绑定 Runner，或在分叉请求体中请求托管主机。若要在不同 Agent 上继续会话，在分叉请求体中添加 `"agent_id"`（GET /v1/agents 列出的内置 Agent）。分叉是更改 Agent 的唯一方式：会话绑定的 Agent 无法原地切换，POST /v1/sessions/{id}/switch-agent 返回 410 Gone。添加 `"side_chat": true`（默认 false）标记分叉为侧边聊天：分叉被标记为 `omnigent.side_chat` 并从 GET /v1/sessions 中省略，因此不出现在 Web UI 侧边栏，仅作为 Workspace-rail 标签页出现。其他行为与其他分叉相同，拥有自己的 Runner 和转录记录。
归档会话	保留历史但将其从默认列表中隐藏： `curl --fail-with-body -X PATCH "http://127.0.0.1:6767/v1/sessions/$SESSION_ID" \` `-H 'Content-Type: application/json' \` `-d '{"archived": true}' \| jq`
恢复会话	使已归档会话再次出现在默认列表中： `curl --fail-with-body -X PATCH "http://127.0.0.1:6767/v1/sessions/$SESSION_ID" \` `-H 'Content-Type: application/json' \` `-d '{"archived": false}' \| jq`
删除会话	永久删除会话及其关联资源。需要所有者级别访问权限： `curl --fail-with-body -X DELETE "http://127.0.0.1:6767/v1/sessions/$SESSION_ID" \| jq`

共享会话：在可访问的共享服务器上使用权限 API 共享会话。你的凭证需要对会话具有 manage 或 owner 访问权限，且服务器的共享设置必须允许请求的访问级别。示例使用 Bearer 令牌身份验证：
SERVER_URL=""
SESSION_ID=""
ACCESS_TOKEN=""

对于特定用户，level 1 表示读取访问，2 表示也可发送消息并与会话交互，3 表示也可管理共享。公共访问始终为只读（level: 1）。

使用场景	示例
与特定用户共享	使用接收方在此服务器上的用户 ID，如账户用户名或 SSO 邮箱。重复此请求会更新其现有访问级别。 `curl --fail-with-body -X PUT "$SERVER_URL/v1/sessions/$SESSION_ID/permissions" \` `-H "Authorization: Bearer $ACCESS_TOKEN" \` `-H 'Content-Type: application/json' \` `-d '{"user_id": "", "level": 1}'`
使会话公开	授予 `__public__` 读取访问权限，使任何能登录服务器的人都能通过 URL 查看会话，而无需单独授权。服务器必须启用公共共享；这不绕过登录。 `curl --fail-with-body -X PUT "$SERVER_URL/v1/sessions/$SESSION_ID/permissions" \` `-H "Authorization: Bearer $ACCESS_TOKEN" \` `-H 'Content-Type: application/json' \` `-d '{"user_id": "__public__", "level": 1}'`
检查当前授权	`curl --fail-with-body "$SERVER_URL/v1/sessions/$SESSION_ID/permissions" \` `-H "Authorization: Bearer $ACCESS_TOKEN" \| jq`
移除公共访问	撤销公共授权；特定用户的授权保留。若要撤销特定用户，将其 URL 编码的用户 ID 替代 `__public__`。 `curl --fail-with-body -X DELETE \` `"$SERVER_URL/v1/sessions/$SESSION_ID/permissions/__public__" \` `-H "Authorization: Bearer $ACCESS_TOKEN"` 向接收方发送会话 URL。特定用户必须使用你授予访问权限的身份登录： `echo "$SERVER_URL/c/$SESSION_ID"`

你也可以让 Agent 通过 sys_session_share 共享会话。在其 Agent 配置中设置 agent_session_sharing: non-public 以允许与用户共享，或设置 agent_session_sharing: public 以也允许公共授权。该工具接受 user_id、level（"read"、"edit" 或 "manage"）以及可选的 session_id（默认指向调用会话）。调用者的权限和服务器共享设置仍然适用。

API 端点参考：服务器暴露这些会话端点。将 {id} 替换为会话 ID。

方法	路径	功能
POST	/v1/sessions	创建会话
GET	/v1/sessions	列出和过滤会话
GET	/v1/sessions/{id}	读取会话详情和状态
PATCH	/v1/sessions/{id}	重命名、配置、归档或取消归档
DELETE	/v1/sessions/{id}	删除会话及其关联资源
POST	/v1/sessions/{id}/fork	将对话历史复制到新会话
GET	/v1/sessions/{id}/items	读取对话历史
GET	/v1/sessions/{id}/child_sessions	列出子会话
GET	/v1/sessions/{id}/stream	通过 SSE 接收实时事件
POST	/v1/sessions/{id}/events	发送消息或中断活动工作
PUT	/v1/sessions/{id}/permissions	授予或更新用户/公共访问
GET	/v1/sessions/{id}/permissions	列出权限授权
DELETE	/v1/sessions/{id}/permissions/{user_id}	撤销权限授权

有关这些及其他端点的更多详情，参见 API Reference。

自动化功能

让 Agent 管理其他会话：你可以要求 Agent 将工作委托给另一个 Omnigent Agent。它使用内置会话工具创建子对话、发送指令并读取结果。这些是 Agent 在会话期间调用的工具；你无需编写 API 客户端即可使用它们。

工具	功能
sys_agent_list	查找可用 Agent 及其 ID。
sys_session_create	从现有 Agent ID 或本地 Agent 配置创建子会话，可选附带第一条消息。
sys_session_send	向子会话发送后续消息，或启动父 Agent 声明的命名子 Agent。
sys_session_list	查找子会话及其他可访问的会话。
sys_session_get_info	检查会话状态、Agent 和工作区。
sys_session_get_history	读取会话的最近消息和工具结果。
sys_session_close	关闭已完成的子会话，同时保留其历史。

例如，在 Omnigent 会话中发送此提示词：
“Find the native Claude Code agent and create a child session titled 'Review tests'. Ask it to review the test coverage in this repository without editing files. Read its findings and summarize the missing coverage for me.”
Agent 可以使用 sys_agent_list 查找 Agent，然后调用 sys_session_create，参数如下：

{
  "agent_id": "",
  "title": "Review tests",
  "message": "Review the test coverage in this repository. Report missing coverage without editing files."
}

返回的 conversation_id 标识子会话。要继续其工作，请要求：
“Ask the 'Review tests' child session to prioritize its findings by risk.”
sys_session_send 只能写入调用会话的直接子节点。对于具有命名子 Agent 的 Agent，重用相同的 Agent 和标题将继续该子对话；不同的标题将启动一个独立的对话。子会话显示在 Web UI 中，你可以跟踪其进度。

使用捆绑技能进行多 Agent 工作流：技能是可复用的指令，指导 Agent 如何使用其工具。可用的技能取决于你运行的 Agent。例如，Polly 捆绑了这些编码工作流：

技能	适用场景
investigate	委托代码探索或调试，然后结合发现结果。
fanout	将独立的实现任务分配给工作器，每个工作器在独立的工作树和 PR 中运行。
cross-review	让不同供应商的 Agent 审查实现，并将阻塞性发现反馈回去以进行修复。

在你的仓库中启动 Polly：
omni polly
然后在提示词中指定技能：
“Use the investigate skill to trace how this repository authenticates API requests. Have separate agents examine the server routes and client code, then summarize the flow with file references. Do not edit files.”
对于准备好审查的更改：
“Use cross-review to review PR #123 against its stated requirements.”
Polly 需要配置好工作器 Harness 及其凭证。技能协调工作；工作器仍需访问仓库和模型。参见 Prompts & Skills 以在 Agent 的 skills/ 目录或项目的 .agents/skills/ 目录中添加你自己的工作流。

调度定期 Agent 运行：当你希望重复相同的工作时，要求 Agent 创建定时任务。Omnigent 提供四个内置工具：

工具	功能
sys_scheduled_task_create	保存提示词、Agent 和定期调度。
sys_scheduled_task_list	列出你的定时任务。
sys_scheduled_task_update	更改任务的提示词、调度或其他设置。
sys_scheduled_task_delete	移除任务，使其不再运行。

例如，替换此提示词中的主机名和仓库路径：
“Create a scheduled task named 'Weekday code review'. Every weekday at 9am Asia/Singapore, run native Claude Code on host 'my-host' in /home/ci/my-repository. Ask it to review the latest commit and report findings without editing files. Look up the agent and host IDs first.”
调度使用此时区的此循环规则：FREQ=WEEKLY;BYDAY=MO,TU,WE,TH,FR;BYHOUR=9;BYMINUTE=0
所选主机在任务触发时必须在线，且 Agent 需要适合无人值守工作的凭证和工具权限。之后你可以要求“Move Weekday code review to 10am”或“Delete Weekday code review”。Agent 使用 list 和 update 或 delete 工具进行修改。完整配置参见 Scheduled Tasks。脚本也可以通过 /v1/scheduled-tasks 管理任务；机器客户端凭证令牌无权访问该端点。

第27章 Omnigent 的上下文感知策略

第29章配置参考 | Omnigent

本篇用到的工具

Omnigent

Omnigent 是一个开源的 AI Agent 框架与元编排层（meta-harness），在 Claude Code、Codex、Cursor、Pi 及自定义 Agent 之上提供统一抽象层。用户无需重写代码即可在多个 Agent 工具间切换或组合，并通过上下文策略、安全沙箱实现成本控制与权限管理，还支持跨设备实时协作共享同一会话。提供 CLI、Web、桌面、移动端及 REST API 多种访问方式。

第28章 编程式使用 Omnigent

本篇用到的工具

评论 (0)

第28章编程式使用 Omnigent