CubeLoopA Pythonic, async-native agent framework.
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, 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 — a langgraph and pi-agent-core alternative
| langgraph | CubeLoop | |
|---|---|---|
| Abstraction | Graph nodes + edges + channels | Plain async functions — run_agent_loop is a while loop |
| Streaming | Callback-based, multiple handler types | async for event in stream — one pattern everywhere |
| Checkpointing | Full snapshot per step; serializes entire message list | Append-only — O(1) DB I/O regardless of conversation length |
| Dependencies | 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 routes and parallelizes |
| Multi-provider | Via langchain chat model adapters | Native Provider protocol — Anthropic, OpenAI built in |
| Middleware | Graph-level middleware on node entry/exit | 8 typed hooks with declarative composition rules |
| Observability | LangSmith / Langfuse integration | Native OpenTelemetry — Tracer, Meter, GenAI semconv, OTLP / JSONL out of the box |
Full comparison: CubeLoop vs LangGraph → · CubeLoop vs pi-agent-core →
Hello, agent.
A single async function loop. One Provider, one AgentTool, and you're streaming.
import asyncio
from cubeloop import Agent, tool
from cubeloop.providers.anthropic import AnthropicProvider
provider = AnthropicProvider(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-6"),
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?"))
Agents
One async loop, fully typed events.
→ Guides / AgentsStreaming
async for event in stream.
→ Guides / StreamingTools
Plain functions, parallel execution.
→ Guides / ToolsProviders
Anthropic, OpenAI, or write your own.
→ Guides / ProvidersCheckpointing
Append-only, O(1) per turn.
→ Guides / CheckpointingHuman-in-the-loop
Pause for confirm, approve, or ask.
→ Guides / Human-in-the-loopMCP
Load remote tools at startup.
→ Guides / MCPTracing
OpenTelemetry, OTLP / JSONL, GenAI semconv.
→ Guides / TracingInstall
pip install cubeloopuv add cubelooppoetry add cubelooppip install cubeloop[sqlite,postgres,mcp,tracing,tracing-otlp]