diva-ai — Diva SDK for Python

Build agents on the Diva platform from Python. A thin client: the agent engine runs server-side on Diva's hosted gateway — you connect with a bearer token (sk-diva-…), the engine never runs locally, and all model traffic goes through the platform.

This is the Python sibling of the TypeScript @diva-ai/sdk and speaks the same gateway wire protocol (session keys are byte-identical, so a Python and a TS client can resume the same server-side conversation).

Status: alpha (0.1.0a1). run / stream / generate, client tools, toolsets, sessions + memory, permissions, hooks, guards, sub-agents, parallel, and stream reconnect are implemented and covered by a live E2E suite. See Feature parity and PORT_PLAN.md.

Install

pip install diva-ai

Requires Python ≥ 3.10. Depends only on websockets and pydantic.

Quickstart

import asyncio
from diva_ai import Agent


async def main() -> None:
    agent = Agent(
        "diva/deepseek/deepseek-v4-flash",
        instructions="You are a concise assistant.",
        api_key="sk-diva-...",          # or set DIVA_API_KEY
    )
    result = await agent.run("What is the capital of France?")
    print(result.text)                  # "Paris."
    print(result.usage)                 # token usage
    await agent.close()


asyncio.run(main())

Point the client at a gateway with DIVA_GATEWAY_URL (e.g. ws://localhost:5002/gateway for a local platform) or pass gateway_url= to Agent. ws:// is allowed only to loopback / private ranges; wss:// always.

Configuration

Option	Source	Notes
API key	api_key= or DIVA_API_KEY	sk-diva-… bearer token
Gateway	gateway_url= or DIVA_GATEWAY_URL	defaults to the hosted endpoint
Model	Agent("diva/<family>/<model>")	namespaced; split into provider+model on the wire

Model refs

Models are namespaced diva/<family>/<model>. The SDK splits this into a provider and model on the wire and the platform routes it to the real backend. A per-turn override is available via run(..., model=...).

Core API

run — one turn

r = await agent.run("Summarize this in one line: ...")
r.text          # the reply
r.usage         # Usage(input_tokens, output_tokens, total_tokens, cache_*)
r.reasoning     # model thinking, when reasoning is enabled (else None)
r.run_id, r.duration_ms, r.stop_reason

stream — token streaming

from diva_ai import DeltaChunk, DoneChunk

async for chunk in agent.stream("Write a haiku about the sea."):
    if isinstance(chunk, DeltaChunk):
        print(chunk.delta, end="", flush=True)
    elif isinstance(chunk, DoneChunk):
        print("\n--", chunk.usage)

If the socket drops mid-stream (a transient close after connect), the SDK transparently resumes via agent.streamEvents replay and fetches the authoritative terminal — a lost terminal fails loud, never silently truncates.

generate — structured output

from pydantic import BaseModel

class Contact(BaseModel):
    name: str
    email: str

res = await agent.generate("Extract: John Smith, john@x.com.", Contact)
res.output          # Contact(name="John Smith", email="john@x.com")
res.attempts        # 1 = one-shot; 2 = repaired on retry
res.repaired

The schema drives the JSON directive and validation; one repair retry runs in a disjoint session so it never pollutes the caller's conversation.

Client tools

Define a tool with a pydantic input schema; the engine calls it and the SDK executes it locally, then returns the result — the model loops until done.

from pydantic import BaseModel
from diva_ai import Agent, tool

class WeatherInput(BaseModel):
    city: str

def get_weather(inp: WeatherInput):
    return {"city": inp.city, "tempC": 21, "sky": "clear"}

agent = Agent(
    "diva/deepseek/deepseek-v4-flash",
    instructions="Answer weather questions by calling get_weather.",
    tools=[tool(name="get_weather", description="Get weather for a city.",
                input_schema=WeatherInput, execute=get_weather)],
)
await agent.run("What is the weather in Lisbon?")

execute may be sync or async. Group related tools with toolsets:

from diva_ai import toolset
agent = Agent(model, toolsets=[toolset("weather", [get_weather_tool])])

An optional display= block gives the dashboard a human label, an icon and a grouping key, without changing anything the model reads:

tool(name="check_order", description="Order status from the ERP",
     input_schema=OrderInput, execute=check_order,
     display={"label": "Check order", "icon": "📦", "category": "ERP"})

Every key is optional; omit one and the platform falls back to the tool's name/description. See Tools.

A tool marked requires_approval=True does not run until an operator approves it in the Diva dashboard — the platform holds the call before it reaches your process, and a denial comes back to the agent as a refusal with the operator's reason:

tool(name="wire_money", description="Send a payment from the company account",
     input_schema=TransferInput, execute=wire_money,
     requires_approval=True)

This is a separate gate from can_use_tool below, and they stack: the operator decides first, your own callback second. See Tools.

Permissions (can_use_tool)

An interactive per-call gate applied client-side before a tool runs (fail-closed):

from diva_ai import Agent, Permissions

async def can_use_tool(name, args):
    if name == "delete_all":
        return {"behavior": "deny", "message": "not allowed"}
    return {"behavior": "allow"}

agent = Agent(model, tools=[...], permissions=Permissions(can_use_tool=can_use_tool,
                                                          allow=["get_weather"]))

permissions.mode / deny target engine built-ins the thin client doesn't expose and raise DivaNotImplementedError — use can_use_tool (+ guard.tool).

Hooks & guards

Lifecycle hooks wrap the turn and tool calls; guards are declarative sugar.

from diva_ai import Agent, Hooks, guard

hooks = Hooks(
    before_agent_start=lambda ev: {"replace": ev["message"].strip()},
    before_reply=lambda ev: {"replace": ev["text"]} if ok(ev["text"]) else {"block": "unsafe"},
    agent_end=lambda ev: log(ev["reply"]),
)

agent = Agent(
    model,
    hooks=hooks,
    guards=[guard.output("password", "secret"),   # hard-block the reply on a match
            guard.tool("rm -rf", tool="exec")],    # soft-block a tool by input
)

Hook outcomes: return None (continue), {"block": reason} (→ DivaGuardTripped), or {"replace": value} (rewrite message/reply/tool-input/tool-output).

Sub-agents (handoff)

Delegate to another agent as a tool — "just a typed tool transfer", no graphs.

from diva_ai import Agent, handoff

qualifier = Agent(model, instructions="Qualify the lead in one line.")
agent = Agent(model, tools=[handoff(qualifier, name="qualifier",
                                    description="Qualify an inbound sales lead")])

Each handoff is an independent, stateless sub-agent turn. Close sub-agents yourself — the parent's close() does not cascade.

Skills

Named instruction/knowledge blocks composed into the system prompt every turn.

from diva_ai import Agent, skill, skill_from_dir

agent = Agent(model, skills=[
    skill(name="objection-handling", description="How to handle pricing pushback",
          body="When the customer says it's too expensive, ..."),
    skill_from_dir("./skills/refunds"),   # reads ./skills/refunds/SKILL.md
])

Skill content is trusted (you author it). Bodies are size-bounded and duplicate names fail loud.

MCP servers

Connect external MCP servers (stdio or HTTP); their tools join the agent as client tools named <server>__<tool>. Requires the mcp extra (pip install 'diva-ai[mcp]'); connections open lazily on the first turn and close with agent.close().

import sys
from diva_ai import Agent, MCP

agent = Agent(
    model,
    mcp=[
        MCP.stdio("filesystem", "npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/data"]),
        MCP.http("weather", "https://mcp.example.com/mcp", headers={"Authorization": "Bearer ..."}),
    ],
)

Sessions & memory

Multi-turn memory two ways:

# Server-side (default): a stable session id keeps history on the platform.
chat = agent.session("user-42")
await chat.run("My name is Ada.")
await chat.run("What is my name?")     # → "Ada"

# Client-side: YOU own the history (in-process or on disk).
from diva_ai import Agent, MemoryStore, FileStore
agent = Agent(model, store=FileStore("./sessions"))   # or MemoryStore()

With a store, prior turns are injected as fenced untrusted-data and the server turn runs stateless. Session keys fold model + instructions + session id and are byte-identical to the TS client, so cross-client resume works.

Flow (slot-filling)

Guide a conversation with a frame: soft guidance (asks + rules) plus a hard GUARANTEE — a terminal action is blocked until its required slots are filled, and tools can be gated on prerequisites.

from diva_ai import Agent, flow

checkout = (
    flow("checkout")
    .slot("address", fill_when={"tool_called": ["set_address"]}, ask="Ask for the shipping address")
    .gate("place_order", require_prior=["set_address"], block_reason="Set the address first.")
    .completion("place_order", requires=["address"])
    .build()
)
agent = Agent(model, tools=[set_address_tool, place_order_tool], flow=checkout)

The interpreter runs client-side over the hooks: it tracks slot beliefs from tool results, blocks a gated tool until its prerequisites are met (up to max_blocks), and hard-blocks the completion action until required slots are filled.

The funnel's shape is reported to the dashboard automatically and drawn on the agent card, read-only. Execution stays here; only the description travels.

What the platform sees

You write no telemetry. With DIVA_API_KEY set, the dashboard shows your agent's model, instructions, skills, guards, funnel and tool catalogue, plus every run with its tool calls, tokens and cost.

Give the agent a readable name with label= — otherwise the dashboard names it from its model plus an identity digest:

agent = Agent("diva/deepseek/deepseek-v4-flash",
              label="Shop support",
              instructions="You are a shop support agent.")

The label is presentation only: the agent's identity is its model plus system prompt, so renaming one keeps its run history in place.

Four things never leave your process: skill bodies, guard blocklists, your tool implementations, and funnel execution. That is why the card shows your instructions= argument rather than the composed system prompt — the composed one has every skill body folded into it.

A funnel also reports, each turn, which of its slots are filled — flags only, never the value a slot was filled with — so a run's trace shows which step the conversation was on.

Full list, and why the line is drawn there: docs/platform-visibility.md.

Parallel fan-out

from diva_ai import parallel

results = await parallel(
    [lambda: agent.run(f"Weather in {c}?") for c in ["Berlin", "Tokyo", "Lima"]],
    concurrency=4,
)
for r in results:
    print(r.value.text if r.status == "fulfilled" else r.reason)

Reasoning

thinking_default (off | minimal | low | medium | high | xhigh | adaptive) maps to each provider's native reasoning control. When on and the model emits reasoning, it is surfaced on result.reasoning, kept separate from result.text.

Errors

All errors subclass DivaError: DivaAuthError (no key/target), DivaHostError (gateway unreachable), DivaRequestError (turn failed/timed out), DivaNotImplementedError (unwired feature), DivaHookError, DivaGuardTripped.

Feature parity

Feature	Status
run / stream / generate	✅
Client tools (inline) + toolsets	✅
Permissions / can_use_tool	✅
Hooks + guards	✅
Sub-agents (handoff)	✅
Sessions + memory (server + MemoryStore/FileStore)	✅
Parallel, thinking levels, observability	✅
Stream reconnect (resumable)	✅
Session-key parity vs TS	✅ (byte-identical)
Skills (skill / skill_from_dir, prepend)	✅
MCP servers (stdio / http)	✅ (diva-ai[mcp])
Flow (slot-filling frames)	✅
Tool display + agent label (D5), wire-identical to TS	✅
Tool requires_approval (D6), wire-identical to TS	✅
Protocol drift guard (mirror of TS npm run check:drift)	✅
Park/resume tool mode	roadmap (engine ISKARIOT_RUN_PARK_RESUME; the hosted gateway is inline)

Testing

pip install -e ".[dev]"
pytest tests/test_unit.py                     # pure logic, no network
pytest tests/test_compat.py                   # protocol backwards compatibility
python scripts/check_protocol_drift.py        # vendored gateway contract guard
DIVA_GATEWAY_URL=ws://localhost:5002/gateway DIVA_API_KEY=sk-diva-... \
  pytest tests/test_e2e_live.py               # live E2E (skipped without env)

check_protocol_drift.py is the mirror of the TypeScript SDK's npm run check:drift: it pins the vendored gateway protocol constants against scripts/protocol-snapshot.json, and re-verifies them against a live engine checkout when DIVA_ENGINE_SRC is set. Python packaging has no script-runner table to alias it behind, so the command above is the whole story — but the same functions are driven in-process by tests/test_protocol_drift.py, so drift reddens a plain pytest run too.

License

Apache-2.0. The SDK is open; the Diva engine and platform are proprietary.