diva-ai 0.1.0a1

Diva SDK for Python (thin client) — build agents on the Diva platform over a hosted gateway (bearer token, no local engine)

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, [e-mail].", Contact)
res.output          # Contact(name="John Smith", email="[e-mail]")
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])])

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.

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)	✅
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
DIVA_GATEWAY_URL=ws://localhost:5002/gateway DIVA_API_KEY=sk-diva-... \
  pytest tests/test_e2e_live.py               # live E2E (skipped without env)

License

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