cubepi · v0.3.0 · alpha

A Pythonic, async-native agent framework.

Plain async functions instead of graph nodes. 3 deps. Append-only checkpointing.

Quick Start →G Q

Why CubePi

	langgraph	CubePi
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	5 typed hooks with declarative composition rules
Observability	LangSmith / Langfuse integration	Events + middleware hooks — bring your own tracing

Hello, agent.

A single async function loop. One Provider, one AgentTool, and you're streaming.

Full quick-start →

import asyncio
from pydantic import BaseModel
from cubepi import Agent, AgentTool, Model
from cubepi.agent.types import AgentToolResult
from cubepi.providers.anthropic import AnthropicProvider
from cubepi.providers.base import TextContent

provider = AnthropicProvider(api_key="sk-...")

class GetWeatherParams(BaseModel):
    city: str

async def get_weather(tool_call_id, params: GetWeatherParams, *, signal=None, on_update=None):
    return AgentToolResult(
        content=[TextContent(text=f"72°F and sunny in {params.city}")]
    )

agent = Agent(
    provider=provider,
    model=Model(id="claude-sonnet-4-5-20250929", provider="anthropic"),
    tools=[AgentTool(
        name="get_weather",
        description="Get current weather for a city",
        parameters=GetWeatherParams,
        execute=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?"))