cubeloop

<p align="center"> <img src="https://raw.githubusercontent.com/cubeplexai/cubeloop/main/website/static/img/brand/cubeloop-social-preview.png" alt="CubeLoop" width="800"> </p> [](https://github.com/cubeplexai/cubeloop/actions/workflows/ci.yml) [](https://codecov.io/gh/cubeplexai/cubeloop) [](https://pypi.org/project/cubeloop/) [](https://pypi.org/project/cubeloop/) [](https://opensource.org/licenses/MIT) [](https://cubeloop.dev) [](https://deepwiki.com/cubeplexai/cubeloop) [](https://skills.sh/cubeplexai/cubeloop) CubeLoop is a Pythonic, async-native agent framework designed for high performance, readability, and production-grade persistence. It provides a leaner alternative to graph-based agent runtimes by modeling agent logic as a linear while loop that developers can easily trace and debug. CubeLoop powers [CubePlex](https://github.com/cubeplexai/cubeplex), a cloud-native platform for managed agents in team workspaces — skills, shared memory, MCP tools, persistent sandboxes, governed access, and self-hosted deploy on Docker Compose or Kubernetes. ## Why CubeLoop | | langgraph | CubeLoop | |---|---|---| | **Abstraction** | Graph nodes + edges + channels — you model your agent as a state machine | Plain async functions — `run_agent_loop` is a while loop you can read in 5 minutes | | **Streaming** | Callback-based, multiple handler types | `async for event in stream` — one pattern everywhere | | **Checkpointing** | Full snapshot per step — serializes entire message list on every channel change | Append-only — writes only new messages, O(1) DB I/O regardless of conversation length | | **Dependencies** | Pulls in langchain-core, langgraph-sdk, and transitive deps | 3 core deps: `pydantic`, `anthropic`, `openai` | | **Tool execution** | Tools are graph nodes with manual wiring | Declare tools as functions, framework handles routing and parallel execution | | **Multi-provider** | Via langchain chat model adapters | Native `Provider` protocol — Anthropic, OpenAI built in, add your own with one class | | **Middleware** | Graph-level middleware on node entry/exit | Agent-level middleware with 8 typed hooks and declarative composition rules | | **Observability** | LangSmith / Langfuse integration, full trace visualization | Native OpenTelemetry — `Tracer`, `Meter`, GenAI semconv, OTLP / JSONL exporters built in | ## Install ```bash pip install cubeloop # Optional extras pip install cubeloop[sqlite] # SQLite checkpointer pip install cubeloop[postgres] # Postgres checkpointer pip install cubeloop[mysql] # MySQL checkpointer pip install cubeloop[mcp] # MCP tool loaders pip install cubeloop[tracing] # OpenTelemetry tracing + metrics pip install cubeloop[tracing-otlp] # Adds the OTLP/HTTP span exporter pip install cubeloop[trace-cli] # `cubeloop trace` terminal viewer ``` Or with [uv](https://github.com/astral-sh/uv): ```bash uv add cubeloop uv add cubeloop[sqlite,postgres,mysql,mcp,tracing] ``` For local Git hooks that mirror CI, install `pre-commit` and enable the repo config: ```bash uvx pre-commit install ``` ## Quick Start ```python import asyncio from cubeloop import Agent, tool from cubeloop.providers.anthropic import AnthropicProvider provider = AnthropicProvider(provider_id="anthropic", api_key="sk-...") @tool async def get_weather(city: str) -> str: "Get current weather for a city." return f"72°F and sunny in {city}" agent = Agent( model=provider.model("claude-sonnet-4-5-20250929"), tools=[get_weather], system_prompt="You are a helpful weather assistant.", ) def on_event(event, signal=None): if event.type == "text_delta": print(event.delta, end="", flush=True) agent.subscribe(on_event) asyncio.run(agent.prompt("What's the weather in Tokyo?")) ``` For a guided tour of the architecture, browse the [DeepWiki for this repo](https://deepwiki.com/cubeplexai/cubeloop) or the [Core Concepts guide](https://cubeloop.dev/docs/getting-started/core-concepts). ## Core Concepts ### Providers Abstract LLM interaction behind a `Provider` protocol. All providers return `MessageStream` — an async iterator of `StreamEvent`s. ```python from cubeloop.providers.anthropic import AnthropicProvider from cubeloop.providers.openai import OpenAIProvider from cubeloop.providers import FauxProvider # Real providers anthropic = AnthropicProvider(provider_id="anthropic", api_key="...") openai = OpenAIProvider(provider_id="openai", api_key="...") # Test provider — no API calls, fully deterministic faux = FauxProvider(provider_id="faux") faux.set_responses(["Hello!", "How can I help?"]) ``` Use `FallbackBoundModel` to chain providers — on a rate limit, outage, or context-length error the next model in the chain is tried automatically: ```python from cubeloop import FallbackBoundModel model = FallbackBoundModel( chain=( anthropic.model("claude-opus-4-8"), # primary openai.model("gpt-5"), # fallback ) ) agent = Agent(model=model, system_prompt="...") ``` ### Tools Decorate an async function with `@tool`: the input schema is derived from the typed parameters, the docstring becomes the description, and the framework handles argument parsing, parallel execution, and error wrapping. ```python from cubeloop import tool @tool async def search(query: str) -> str: "Search the web." return f"Results for: {query}" ``` Need a shared params model, dynamic construction, or `execution_mode`? The longhand `AgentTool(...)` is equivalent and fully supported: ```python from pydantic import BaseModel from cubeloop import AgentTool from cubeloop.agent.types import AgentToolResult from cubeloop.providers.base import TextContent class SearchParams(BaseModel): query: str async def execute(tool_call_id, params: SearchParams, *, signal=None, on_update=None): return AgentToolResult(content=[TextContent(text=f"Results for: {params.query}")]) search = AgentTool( name="search", description="Search the web", parameters=SearchParams, execute=execute, execution_mode="parallel", # or "sequential" ) ``` ### Middleware Composable hooks that modify behavior without touching the core loop: ```python from cubeloop import Middleware, compose_middleware from cubeloop.agent.types import BeforeToolCallResult class LoggingMiddleware(Middleware): async def transform_context(self, messages, *, ctx, signal=None): print(f"Context has {len(messages)} messages") return messages class SafetyMiddleware(Middleware): async def before_tool_call(self, ctx, *, signal=None): if ctx.tool_call.name == "dangerous_tool": return BeforeToolCallResult(block=True, reason="Blocked by policy") return None hooks = compose_middleware([LoggingMiddleware(), SafetyMiddleware()]) ``` **Composition rules:** | Hook | Rule | |------|------| | `transform_context` | Chained — each receives previous result | | `convert_to_llm` | Last implementation wins | | `resolve_tool_call` | First non-`None` rewrite wins (short-circuits) | | `before_tool_call` | Any block stops execution | | `after_tool_call` | Later overrides earlier | | `transform_system_prompt` | Chained — each receives previous result | | `after_model_response` | Returns `TurnAction`; last decision wins, messages concatenate | | `should_stop_after_turn` | Any true stops | | `on_run_end` | Messages concatenate; non-empty result triggers one extra model turn | ### Deferred Tool Groups When an agent has access to many MCP servers, their combined tool schemas can consume significant context. Deferred tool groups hide schemas by default and let the model expand them on demand: ```python from cubeloop import Agent from cubeloop.deferred import DeferredToolGroup # load_github_tools is a zero-arg async callable returning list[AgentTool] # (e.g. wrap load_mcp_tools_stdio(...).tools — see the website guide). github_group = DeferredToolGroup( group_id="mcp:github", display_name="GitHub", description="Issues, PRs, repos, code search", tool_names=["create_issue", "search_repos", "create_pr", "list_comments"], loader=load_github_tools, ) agent = Agent( model=provider.model("claude-sonnet-4-6"), tools=[get_weather], # always-available tools deferred_tool_groups=[github_group], # hidden until requested ) ``` The model sees a compact catalog in the system prompt instead of full schemas: ``` # Deferred tool groups - `mcp:github` — GitHub: Issues, PRs, repos, code search (4 tools) create_issue, search_repos, create_pr, list_comments ``` When the model needs a group, it calls the built-in `load_tools` tool: ``` load_tools(group_id="mcp:github") # load all load_tools(group_id="mcp:github", tool_names=["create_issue"]) # or just one ``` The loader is called once per group per run; subsequent selective loads filter from the cached result. With the default `dispatch` strategy, `load_tools` returns the full schemas in its tool result and loaded tools are invoked via the `deferred_tool_call` dispatcher — the tools array and system prompt stay byte-stable, so loading never invalidates the prompt cache. The v1 behavior (native injection into the model-visible tools array) is available with `deferred_tool_strategy="inject"`. For advanced use (custom catalog header, cross-run replay), construct `DeferredToolsMiddleware` directly: ```python from cubeloop.deferred import DeferredToolsMiddleware # Replay expansion state from a previous run (strategy is required and # must match the middleware's strategy) resumed = await DeferredToolsMiddleware.prepare_resumed_state( groups=all_groups, expanded=saved_extra["expanded_groups"], strategy="dispatch", ) agent = Agent( model=model, tools=[*builtins, *resumed.pre_loaded_tools], deferred_tool_groups=resumed.remaining_groups, ) ``` ### Checkpointer Persist conversation state with append-only semantics: ```python from cubeloop.checkpointer import ( MemoryCheckpointer, SQLiteCheckpointer, PostgresCheckpointer, MySQLCheckpointer, ) # In-memory for dev/test cp = MemoryCheckpointer() # SQLite for lightweight persistence async with SQLiteCheckpointer("agent.db") as cp: agent = Agent(model=model, checkpointer=cp, thread_id="conv-1") # Postgres for production async with PostgresCheckpointer("postgresql://...") as cp: agent = Agent(model=model, checkpointer=cp, thread_id="conv-1") # MySQL for production async with MySQLCheckpointer("mysql://...") as cp: agent = Agent(model=model, checkpointer=cp, thread_id="conv-1") ``` Postgres and MySQL never issue DDL at runtime — your app owns the schema via Alembic. See the host-integration runbooks ([Postgres](cubeloop/checkpointer/postgres/README.md) · [MySQL](cubeloop/checkpointer/mysql/README.md)) and the runnable [`examples/`](examples/). ### FauxProvider for Testing Ship your agent tests without API keys: ```python from cubeloop.providers import FauxProvider, faux_text, faux_tool_call, faux_assistant_message provider = FauxProvider(provider_id="faux") provider.set_responses([ faux_assistant_message([ faux_tool_call("search", {"query": "python"}), ]), faux_assistant_message("Here are the results..."), ]) agent = Agent(model=provider.model("test"), tools=[search_tool]) agent.subscribe(lambda event, signal=None: None) # subscribe before prompt to receive events await agent.prompt("Search for python") # Streams realistic deltas — content_block_start, text_delta, etc. ``` ### Tracing Attach a `Tracer` and every agent run produces OpenTelemetry spans aligned with the [GenAI Semantic Conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/) — ingestible by Jaeger, Tempo, Honeycomb, Datadog, AWS X-Ray, or any OTLP-compatible backend without custom instrumentation: ```python from cubeloop.tracing import Tracer, tracing_context from cubeloop.tracing.exporters import JsonlSpanExporter async with ( Tracer( service_name="my-bot", agent_name="assistant", exporters=[JsonlSpanExporter(directory="./cubeloop-traces")], ) as tracer, tracer.attached(agent), ): with tracing_context(tags=["beta-arm"], metadata={"user_id": "u-42"}): await agent.prompt("Hello.") # On exit: detach (closes any cancelled-run spans + flush) + tracer shutdown. ``` Span tree per run: ``` trace └── invoke_agent 14425.8ms [0x1cd97cdb] ← one per agent.prompt() ├── cubeloop.turn 1283.1ms [0x5cfda93e] ← one per LLM round-trip │ ├── chat deepseek-v4-flash 1208.7ms tok 6845/68 [0x0d130229] │ └── execute_tool subagent 9610.2ms subagent [0x38bdd10a] │ └── invoke_agent 9601.0ms [0x8094f99b] ← subagent run, nested │ └── cubeloop.turn 9598.4ms [0x57c5cfc7] │ ├── chat deepseek-v4-flash 1190.3ms [0x8205ca6b] │ └── execute_tool web_search 6500.2ms web_search [0xca4e59fc] └── cubeloop.turn 491.9ms ERROR [0xce25f242] └── chat deepseek-v4-flash 427.2ms ERROR [0x0bff68ec] └── error: Error code: 400 - ... `tool_use` ids were found without `tool_result` blocks immediately after: call_01_... ``` No prompts / model outputs are recorded by default. Opt in with `Tracer(record_content=True)` plus a `redact` callback for PII. Pair with `Meter(...)` for `gen_ai.client.operation.duration` / TTFC / token-usage histograms. Full guide: https://cubeloop.dev/docs/guides/tracing/overview #### Inspecting traces from the terminal With `JsonlSpanExporter` writing to `./cubeloop-traces`, inspect runs with the `cubeloop trace` CLI (install the extra: `pip install cubeloop[trace-cli]`). All subcommands take `--dir` (default `./cubeloop-traces`): ```bash cubeloop trace ls # recent runs, newest first; the `input` # column shows the user message + `status` cubeloop trace view <run_id> # render a run as a tree; errors print inline # under the failing span (no flag needed). # A unique run-id PREFIX is enough. cubeloop trace view <run> --content # also expand prompts / tool args / results cubeloop trace view <run> -v # expand ALL span attributes (verbose) cubeloop trace follow <run_id> # stream spans live as they complete cubeloop trace stats --by model # token / latency / error aggregates cubeloop trace stats --by tool --since 2026-01-01 ``` Typical debugging flow: `ls` (find the run by its `input`), then `view <prefix>` and read the inline `error:` line under any `ERROR` span. Need content only recorded with `Tracer(record_content=True)`. **Token / cache fields.** The recorder reconciles to the GenAI semconv, so `gen_ai.usage.input_tokens` is the **inclusive** total prompt (`input + cache_read + cache_creation`) and `gen_ai.usage.cache_read.input_tokens` is a subset of it. From trace fields, cache hit rate is `cache_read / input_tokens` (≤ 100%) — do **not** add `cache_read` to the denominator. Coding agents debugging cubeloop/consumer apps can install the [`cubeloop-trace` skill](skills/cubeloop-trace/SKILL.md): ```bash npx skills add cubeplexai/cubeloop@cubeloop-trace -a claude-code ``` ## AI Agents Two skills are available for coding agents (Claude Code, Cursor, Codex, …) working with this repo: | Skill | Install | Purpose | |-------|---------|---------| | `cubeloop` | `npx skills add cubeplexai/cubeloop@cubeloop -a claude-code` | Build agents — API reference, tools, middleware, checkpointing, MCP, HITL | | `cubeloop-trace` | `npx skills add cubeplexai/cubeloop@cubeloop-trace -a claude-code` | Debug runs — inspect OTel spans, token counts, tool results, streaming failures | ## Requirements - Python >= 3.11 - Core: `pydantic`, `anthropic`, `openai` - Optional: `aiosqlite` (`[sqlite]`), `asyncpg` + `sqlalchemy` + `msgpack` (`[postgres]`), `aiomysql` + `sqlalchemy` + `msgpack` + `cryptography` (`[mysql]`), `mcp` (`[mcp]`), `opentelemetry-sdk` (`[tracing]`), `opentelemetry-exporter-otlp-proto-http` (`[tracing-otlp]`), `rich` (`[trace-cli]`) ## Credits Architecture inspired by pi-agent-core (TypeScript) and LangGraph; CubeLoop is an independent Python reimplementation with Pydantic v2, asyncio-native primitives, and built-in checkpointing. ## License MIT