关于 🚀 MCPOmni Connect - Universal Gateway to MCP Servers
MCPOmni Connect(OmniCoreAgent)是一个 Python Agent 运行时,覆盖循环、工具、记忆、工作区、子 Agent 以及策略、沙箱、持久化运行、预算和运行记录等能力。可通过 Python、命令行或 HTTP 服务使用。
主要功能
- 01通过 stdio、SSE 或 streamable HTTP 接入 MCP 服务
- 02每次操作前执行策略检查(允许、询问或拒绝)
- 03沙箱中执行命令,并支持多种沙箱提供方
- 04崩溃后可恢复的持久化运行
- 05预算限制与每步运行记录
适用场景
- 01构建需要审批和预算控制的自动化 Agent
- 02在 CI 中以命令行方式无人值守运行 Agent
- 03通过 HTTP 服务向应用提供 Agent 运行能力
MCP 相关
该运行时可通过 stdio、SSE 或 streamable HTTP 连接多个 MCP 服务,并将其工具纳入统一目录供 Agent 调用。
详细介绍
让你的 Agent 承担实际工作,同时保持掌控。
每项操作都经过检查才会执行。每次运行都能在崩溃后恢复,不会悄悄重复任何操作。每一步都有记录。
文档 · 快速入门 · 实用指南 · 验证 · 对比说明 · 面向你的编码 Agent · 已知问题 · 咨询 AI
模型并不是 Agent。让模型能在应用中发挥作用的是它周围的运行时:循环、工具、记忆、它操作的文件,以及——当 Agent 能够执行实际操作时——规定它可以做什么的策略、运行其代码的沙箱、崩溃后仍能保留的记录、阻止它超支的预算,以及事后供人查看的证据。
OmniCoreAgent 是一个面向 Python 的 Agent 运行时。它包含模型周围的 运行框架、循环、工具、上下文、工作区和子 Agent,并提供实际运行 Agent 所需的能力:对每项操作制定策略、以沙箱作为命令执行边界、崩溃后可恢复的运行、预算,以及每一步的记录。你可以通过 Python、命令行或 HTTP 服务使用它:从第一个脚本到受管控的后台工作进程,再到基准测试,都使用同一个 Agent 对象。
| 层 | 内容 | 提供的能力 |
|---|---|---|
| 运行框架 | 模型周围的循环 | 工具(自有工具和 MCP 工具)、上下文与记忆、工作区、代码模式、技能、子 Agent 和工作进程配置 |
| 控制 | 每项操作执行前作出决定 | 策略(允许、询问人员或拒绝)、审批、预算,以及作为命令执行边界的沙箱 |
| 持久性 | 每次运行都会保留的记录 | 暂停与恢复、崩溃后恢复且不重做工作、接管已退出进程的任务 |
| 运行记录 | 每次运行的证据 | 每次运行一条轨迹,从头到尾清晰可读,默认私密,可导出到指定位置 |
了解各部分如何协同工作 将通过一次运行介绍全部四层。
安装
pip install omnicoreagent # Python 3.12–3.14; check with python --version
export LLM_API_KEY=your_api_key # the key for the provider in model_config
在 Python 3.10 或 3.11 上,安装会停止并说明原因。
快速开始
import asyncio
from omnicoreagent import OmniCoreAgent, ToolRegistry
tools = ToolRegistry()
@tools.register_tool("lookup_order")
def lookup_order(order_id: str) -> dict:
"""Look an order up in the application's own store."""
return {"order_id": order_id, "status": "shipped", "carrier": "DHL"}
agent = OmniCoreAgent(
name="support",
system_instruction="You answer questions about orders, using the tools. Answer in plain text.",
model_config={"provider": "openai", "model": "gpt-5.6-terra"},
local_tools=tools,
)
async def main():
result = await agent.run("Where is order 1042?", session_id="customer-7")
print(result["response"])
# Every run is evidence: each step, what the model asked for,
# and what it received back.
trajectory = await agent.get_trajectory(result["trace_id"])
for step in trajectory["steps"]:
for call in step["tool_calls"]:
print(step["step"], call["tool_name"], call["arguments"], "->", call["observation"]["content"])
# What the run turned out to be worth, whenever that is known ...
await agent.record_outcome(result["run_id"], source="support-lead", reward=1.0, label="resolved")
# ... and the run as one record for an evaluator or a trainer.
[record] = await agent.training_records(run_id=result["run_id"])
print(record["outcomes"])
await agent.cleanup()
asyncio.run(main())
输出内容:
Order 1042 has shipped via DHL.
1 lookup_order {'order_id': '1042'} -> {"tool_name": "lookup_order", "args": {"order_id": "1042"}, "status": "success", "data": {"order_id": "1042", "status": "shipped", "carrier": "DHL"}, "message": null}
[{'outcome_id': 'outcome_57a6d81724c44f51b08c540fe83f4694', 'reward': 1.0, 'label': 'resolved', 'source': 'support-lead', 'detail': {}, 'recorded_at': '2026-09-25T18:00:36.136396+00:00'}]
这就是完整的循环:模型调用工具(独立调用会在一个批次中运行),结果以结构化观察的形式返回,会话会记住内容,文件会保存到工作区,注入防护机制会进行监控,运行过程也会被记录。所有操作都遵循默认策略;该策略允许使用相应工具,并会拒绝原始机密、主机 Shell 命令和不受限制的网络访问。只有在需要时,才需要添加下面的内容。
通过一个 model_config,即可搭配 OpenAI、Anthropic、Gemini、Groq、DeepSeek、Mistral、Azure、OpenRouter 和 Ollama 使用(模型)。
每次运行都有证据
Agent 的最终回答会省略大部分过程。运行时会为每次运行保留其余内容,并提供完整、可读的记录:
- 模型在每一步看到的内容——消息和提供给它的工具——以及它收到的内容。两者并不总是与工具返回的内容相同:大型结果会保存到文件中,并向模型提供预览;运行轨迹会同时显示两者。
- 稍后才出现的结果。 已合并的拉取请求、评审、通过的 CI 运行:
record_outcome会将其关联到完成相关工作的运行。 - 训练记录。
training_records会将每次已完成的运行作为一条记录返回,包含发送给模型的内容、模型生成的内容、工具的回复、为其提供服务的策略、各项总计及其结果,供评估器或训练器使用。 - 它自己的存储,以及你的存储。 运行时会自行保留证据,并将其导出到 OTLP、LangSmith、Opik 或 JSONL。它持有的凭据绝不会交给模型,也不会写入记录。
从脚本到 CI,再到基准测试
同一个 Agent 文件可以在无头模式下运行——输入一条指令,得到一个终止状态、退出码,以及写入磁盘的结果和运行轨迹——用于 CI 和脚本:
omnicoreagent run --agent agent.py \
--instruction "Fix the failing test in tests/test_orders.py" \
--approval-mode deny --timeout 900 --output-dir ./out
在 Harbor 上,Terminal-Bench 也基于这个框架运行:Agent 会安装到每个任务的容器中,每次试验都会报告奖励、成本、令牌数和 Harbor 自有格式的运行轨迹,并与其他 Agent 的结果并列展示。
pip install "omnicoreagent[harbor]"
omnicoreagent harbor doctor -m gpt-5.6-terra
omnicoreagent harbor run -d [email protected] -m gpt-5.6-terra -n 4
omnicoreagent harbor results jobs
生产环境所需功能及其位置
| 需求 | 运行时的行为 | 阅读 |
|---|---|---|
| 工具 | 通过统一目录管理你的 Python 函数和 MCP 服务(stdio、SSE、streamable HTTP、OAuth);支持并行批处理;通过调用签名检测循环;支持大型工具集的工具检索。 | 本地工具, MCP |
| 代码模式 | 提供 run_code 工具:模型编写一段简短的 Python 程序来调用你的工具、循环执行并计算,然后在 Monty 中运行——程序内的每次调用都会受到管控并被追踪;需要审批的调用会暂停程序本身。 | 代码模式 |
| 管控 | 对 Agent 拥有的每种能力应用策略——允许、询问、拒绝:每个工具、每个 MCP 服务、沙箱、网络、委派和后台运行。ask 会暂停运行,等待人员处理。Shell 命令可通过你编写的前缀规则进行匹配(在 git push 前询问;拒绝 rm);使用此类规则时,规则绝不会允许无法完整读取的命令:系统会对此询问(在 strict 模式下拒绝)。策略经过哈希处理,因此无法在运行时扩大权限。 | 安全模型, 策略, 审批 |
| 执行 | 提供 execute 工具,命令在沙箱中运行——可使用 Docker、E2B、Modal、Daytona、Vercel、自建沙箱,或在容器已构成隔离边界时使用 local;除非策略允许,否则不提供网络访问,也无法访问进程中的密钥和令牌;工作区会桥接进出沙箱。沙箱就是隔离边界:如果清单中设置的所选提供方无法强制执行隔离(对于内置提供方,会在构建 Agent 时检查),就会被拒绝,而不会先获批再被忽略;local 不是沙箱。沙箱终止时会报告并替换。 | 执行, 提供方 |
| 持久化运行 | 每次运行都有记录:其步骤、工具调用和审批。运行会因等待审批或额度补充而暂停,并从停止处恢复;使用持久化内存存储时,进程终止的运行会从检查点继续;被中断的调用绝不会被悄然重复,除非其工具在调用时及当前均为幂等。 | 持久化运行 |
| 预算 | 可按请求、会话、Agent 或应用设置美元、令牌、调用次数和沙箱运行秒数限制,也可按日或月设置;每次模型调用都会在发出前定价并预留预算;预算耗尽的运行会等待人员处理。 | 预算 |
| 记忆和上下文 | 会话记忆可存储在内存、Redis、Postgres/SQL 或 MongoDB 中;每次模型调用前都会管理上下文;大型工具输出会转存到工作区文件。 | 记忆, 上下文, 存储和扩展 |
| 子 Agent | 由主 Agent 启动的工作 Agent 各自拥有与父级关联的追踪记录,并使用主 Agent 的同一份预算。使用工作 Agent 配置时,主 Agent 会为每项任务选择工作 Agent 类型,包括其模型、推理强度、工具和更严格的规则。工作 Agent 请求审批或预算时,主 Agent 会暂停。 | 子 Agent |
| 后台工作 | 使用持久化任务存储(Redis、MongoDB、SQL)的计划任务和手动任务,支持租约、重试、重启后恢复,并确保每个任务同一时间只运行一次。 | 后台 Agent |
| 遥测 | 每次运行都有一条端到端可读的追踪记录——涵盖每次模型调用、工具调用、沙箱命令、审批和预算决策——默认记录完整信息(capture: "default" 不记录模型提示词);记录中的个人数据会被脱敏(运行本身不会);可导出到 OTLP、LangSmith、Opik 或 JSONL。 | 查看一次运行, 遥测, 隐私 |
| 服务 | omniserve run --agent agent.py:为运行、审批、预算、后台任务和追踪记录提供 REST 与 SSE;支持身份验证、速率限制和 Prometheus 指标;提供每个进程的接入限制,可在同一共享存储上通过负载均衡器部署多个副本,并由服务端恢复死进程遗留的运行;还可在 API 旁托管你自己的页面。 | OmniServe |
| 一个受治理的 Agent,只需一份配置: |
agent = OmniCoreAgent(
name="steward",
system_instruction="...",
model_config={"provider": "openai", "model": "gpt-5.6-terra"},
mcp_tools=[{"name": "github", "transport_type": "streamable_http", "url": "https://api.githubcopilot.com/mcp/",
"headers": {"Authorization": "Bearer ..."}}],
agent_config={
"governance_config": {
"enabled": True,
"policy": {"name": "steward", "mode": "strict", "rules": {
"allow": [{"rule_id": "read", "capability": "tool.mcp.call",
"target": {"mcp_server": "github", "tool_name": "get_file_contents"}},
{"rule_id": "sandbox", "capability": "sandbox.execute"},
{"rule_id": "commands", "capability": "process.exec",
"constraints": {"sandbox_required": True}},
# The manifest asks for the network; a strict policy must allow it.
{"rule_id": "network", "capability": "sandbox.network.configure"}],
"ask": [{"rule_id": "pr", "capability": "tool.mcp.call",
"target": {"mcp_server": "github", "tool_name": "create_pull_request"}}],
"deny": [{"rule_id": "merge", "capability": "tool.mcp.call",
"target": {"mcp_server": "github", "tool_name": "merge_pull_request"}}],
}},
"budgets": {"application_id": "steward",
"application": [{"meter": "model_cost_usd", "limit": 5.0, "window": "day"}],
"request": [{"meter": "model_cost_usd", "limit": 1.0}]},
"sandbox_config": {"provider": "e2b"},
"sandbox_manifest": {"network_policy": {"default": "allow"}},
},
},
telemetry_config={"capture": "full"},
)
用实证说话,而非罗列功能
运行时经过验证:让它处理真实且棘手的工作,并从外部对它施压。本仓库的仓库维护 Agent是第一个应用(apps/steward/):它是在服务器上运行的后台 Agent,会在沙箱中复现失败的测试,在获得人员批准后修复问题,提交附有自身追踪记录链接的拉取请求,将自身故障分类为待办工作,并按计划运行。Harbor 上的试验是第二个应用:这些任务经过专门设计,只能通过它们要验证的功能才能完成;另一些任务则专门设计成会出错。每项试验都会通过检查结束过程来评估,而不只看奖励。支持服务台(apps/support_desk/)是第三个应用:一群客户申请退款,由人员审批;系统逐步扩展到 100 个并发用户,持续承压 30 分钟,并在 11 种情况下接受故障测试(包括退款处理期间服务器和数据库被终止、模型提供商故障),确保每笔退款只计算一次。开发过程中发现的问题——69 项发现,每项都说明了发生了什么、原因是什么,以及如何修复——详见生产验证报告。
本文还逐项比较了它与 OpenAI Agents SDK、LangGraph、Pydantic AI、Claude Agent SDK 和 CrewAI 的能力,包括这些框架更强的方面;每项能力均有来源。
只安装你要用的内容
pip install "omnicoreagent[serve]" # OmniServe REST/SSE
pip install "omnicoreagent[docker]" # Docker sandboxes; e2b, modal, daytona, vercel likewise
pip install "omnicoreagent[redis]" # Redis memory and task store; postgres, mongodb likewise
pip install "omnicoreagent[s3]" # S3 / R2 workspace storage
pip install "omnicoreagent[tokenizer]" # token-exact context and budget estimates
pip install "omnicoreagent[otel]" # OTLP export; langsmith, opik likewise
pip install "omnicoreagent[codemode]" # code mode, in Monty
pip install "omnicoreagent[harbor]" # Harbor and Terminal-Bench trials
pip install "omnicoreagent[all]" # every extra above except harbor
使用 AI 编码 Agent?
请让它查看 AGENTS.md——这是本仓库的 Agent 导览,说明各部分的位置、如何运行测试,以及哪些页面介绍哪些内容。文档提供 llms.txt、复制为 Markdown 和 MCP 服务,也是出于同样的考虑(使用 AI 工具阅读文档)。
指南
快速入门 · 实际应用 · 后台 Agent · OmniServe · 生产环境
开发
git clone https://github.com/omnirexflora-labs/omnicoreagent.git && cd omnicoreagent
uv venv && source .venv/bin/activate
uv sync --all-extras --all-groups --locked
pytest tests/
请参阅 CONTRIBUTING.md。设计说明和计划位于
engineering/。
许可证与作者
MIT——详见 LICENSE。由 Abiola Adeshina (@abiorhmangana)开发。 基于 LiteLLM、FastAPI 和 Pydantic 构建。
评论
一个开放协议,规定了 AI 应用如何以统一方式连接外部工具与数据源。客户端负责与模型对话,MCP 服务负责提供工具、资源和提示模板,两者通过标准消息通信,同一个服务可以被不同客户端复用。