cubepi · v0.13.0

CubePiA Pythonic, async-native agent framework.

Name: CubePi
Author: CubePi

CubePi 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.

Quick Start →G Q

Why CubePi — a langgraph and pi-agent-core alternative

	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	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: CubePi vs LangGraph → · CubePi vs pi-agent-core →

Hello, agent.

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

Full quick-start →

import asyncio
from cubepi import Agent, tool
from cubepi.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?"))