mcpField
🚀
🚀 MCPOmni Connect - Universal Gateway to MCP Servers

🚀 MCPOmni Connect - Universal Gateway to MCP Servers

github.com/Abiorh001框架与 SDK2523 周前更新

关于 🚀 MCPOmni Connect - Universal Gateway to MCP Servers

MCPOmni Connect(OmniCoreAgent)是一个 Python Agent 运行时,覆盖循环、工具、记忆、工作区、子 Agent 以及策略、沙箱、持久化运行、预算和运行记录等能力。可通过 Python、命令行或 HTTP 服务使用。

主要功能

  1. 01通过 stdio、SSE 或 streamable HTTP 接入 MCP 服务
  2. 02每次操作前执行策略检查(允许、询问或拒绝)
  3. 03沙箱中执行命令,并支持多种沙箱提供方
  4. 04崩溃后可恢复的持久化运行
  5. 05预算限制与每步运行记录

适用场景

  1. 01构建需要审批和预算控制的自动化 Agent
  2. 02在 CI 中以命令行方式无人值守运行 Agent
  3. 03通过 HTTP 服务向应用提供 Agent 运行能力

MCP 相关

该运行时可通过 stdio、SSE 或 streamable HTTP 连接多个 MCP 服务,并将其工具纳入统一目录供 Agent 调用。

详细介绍

OmniCoreAgent

让你的 Agent 承担实际工作,同时保持掌控。
每项操作都经过检查才会执行。每次运行都能在崩溃后恢复,不会悄悄重复任何操作。每一步都有记录。

PyPI 下载量 PyPI 版本 Python 版本 许可证

文档 · 快速入门 · 实用指南 · 验证 · 对比说明 · 面向你的编码 Agent · 已知问题 · 咨询 AI


模型并不是 Agent。让模型能在应用中发挥作用的是它周围的运行时:循环、工具、记忆、它操作的文件,以及——当 Agent 能够执行实际操作时——规定它可以做什么的策略、运行其代码的沙箱、崩溃后仍能保留的记录、阻止它超支的预算,以及事后供人查看的证据。

OmniCoreAgent 是一个面向 Python 的 Agent 运行时。它包含模型周围的 运行框架、循环、工具、上下文、工作区和子 Agent,并提供实际运行 Agent 所需的能力:对每项操作制定策略、以沙箱作为命令执行边界、崩溃后可恢复的运行、预算,以及每一步的记录。你可以通过 Python、命令行或 HTTP 服务使用它:从第一个脚本到受管控的后台工作进程,再到基准测试,都使用同一个 Agent 对象。

层内容提供的能力
运行框架模型周围的循环工具(自有工具和 MCP 工具)、上下文与记忆、工作区、代码模式、技能、子 Agent 和工作进程配置
控制每项操作执行前作出决定策略(允许、询问人员或拒绝)、审批、预算,以及作为命令执行边界的沙箱
持久性每次运行都会保留的记录暂停与恢复、崩溃后恢复且不重做工作、接管已退出进程的任务
运行记录每次运行的证据每次运行一条轨迹,从头到尾清晰可读,默认私密,可导出到指定位置

了解各部分如何协同工作 将通过一次运行介绍全部四层。

运行流程:你的应用调用 OmniCoreAgent,由它通过策略、预算和沙箱来管控每项操作,与模型和工具交互,并保留运行证据——轨迹、结果和训练记录——并导出到 OTLP、LangSmith、Opik、JSONL 和 Harbor。

安装

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

(无头运行,Harbor)

生产环境所需功能及其位置

需求运行时的行为阅读
工具通过统一目录管理你的 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 构建。

评论

支持 Markdown

第一次接触 MCP?

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

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