coactra

Component-based composition library for agentic systems. Coactra is Team-first: Team owns agent registration, policy, and model resolution, while Agent remains a thin runtime shell over pydantic-ai.

pip install "coactra[agent]"

Common extras:

pip install "coactra[agent-gateway,oauth]"  # MCP gateway and OAuth
pip install "coactra[graphiti]"
pip install "coactra[mem0]"

Small Surface

from coactra import Team, Scope, Policy, Skill

Submodule helpers when wiring backends:

from coactra.agent import MCPServer

Use the root surface for application code. Use lower-level modules only for backends and host runtime wiring:

• coactra.agent for event/runtime types and outbound A2A transport (coactra.agent.backends)
• coactra.team.directory for org/member/seat persistence
• coactra.memory and coactra.workspace for backend integration

Threads

A Thread is one conversation: a Scope-keyed message log that any agent in the same tenant can continue. Pass it to send()/run() — each finished run replaces the thread's messages with the full history, so handoffs keep context.

from coactra import InMemoryThreadStore, Scope, Thread

store = InMemoryThreadStore()
scope = Scope(tenant_id="local", session_id="chat-42")
thread = await store.load(scope) or Thread(scope=scope)

await agent.run("Where did we leave off?", thread=thread)
await reviewer.run("Continue the plan.", thread=thread)  # same history, other agent

await store.save(thread)  # persistence is explicit; hosts implement ThreadStore

Thread messages are opaque runtime objects. For JSON persistence with the default pydantic-ai runtime, use dump_messages / load_messages from coactra.agent.runtime. Threads are the verbatim conversation log; memory stays the long-term extracted store — the two are independent.

Governance

Gate dangerous tools behind human approval, cap usage per run, and cancel runs:

from coactra import AgentSpec, Limits, Scope, Thread

spec = AgentSpec(
    name="coder",
    model="openai:gpt-5",
    tools=[shell, read_file],
    require_approval=("shell",),          # these tools pause for a human
    limits=Limits(total_tokens=200_000),  # per-run cap; send(limits=...) overrides
)

thread = Thread(scope=Scope(tenant_id="local", session_id="job-7"))
result = await (await agent.send("fix the build", thread=thread)).wait()
if result.status == "paused":
    # show result.pending (tool name + args) to a human, then:
    approvals = {p.id: True for p in result.pending}  # False denies
    result = await (await agent.resume(approvals, thread=thread)).wait()

run = await agent.send("long job", thread=thread)
run.cancel()  # interrupts wait(); streams stop at the next event boundary

A Policy also governs tools: when policy.check returns requires_approval for action="tool:call", that tool is gated; deny removes it from the agent. The pause lives entirely in the thread's messages, so any same-tenant agent can resume it — even after a process restart, via your ThreadStore.

Bring Existing Pieces

import asyncio
import os

from coactra import Team
from coactra.agent import MCPServer


async def main() -> None:
    team = Team.local(
        tenant_id="local",
        namespace="default",
        capability="existing-stack",
        model="openai/qwen3.6-plus",
        api_base="https://opencode.ai/zen/go/v1",
        api_key=os.environ["OC_KEY"],
    )
    agent = await team.add_agent(
        name="existing-stack-agent",
        tools=[MCPServer(url="https://tools.example/mcp", name="tools")],
        memory="inprocess",
        workspace="./desk",
    )
    try:
        print(await agent.run("Use my existing stack"))
    finally:
        await agent.aclose()


asyncio.run(main())

See examples/acceptance/ for the checked examples that define the alpha acceptance path.