Python SDK

pip install dobby-ai-sdk

from dobby_sdk import DobbyClient

client = DobbyClient(
    api_key="gk_user_...",
    org_id="org_...",
    tenant_id="tenant_...",
)

# LLM Completions (OpenAI-compatible)
response = client.chat.completions.create(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": "Explain AI agents in 3 sentences"}],
    max_tokens=200,
)
print(response.choices[0].message.content)

# Streaming
stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a haiku about agents"}],
    stream=True,
)
for chunk in stream:
    content = chunk.choices[0].delta.content or ""
    print(content, end="")

from dobby_sdk import DobbyClient, DobbyPolicyDenied

client = DobbyClient(
    api_key="sk_live_...",
    org_id="org_...",
    tenant_id="tenant_...",
)

# Ask Dobby BEFORE your agent runs a risky action.
# enforce() raises DobbyPolicyDenied if the policy blocks it (fail-closed).
try:
    client.controls.enforce("send_external_email", arguments={"to": addr})
    send_external_email(addr)            # runs ONLY if the policy allowed it

except DobbyPolicyDenied as denied:
    # Your own code is the enforcement point: stop the action,
    # then report that you HONORED the deny.
    client.controls.report(
        denied.verdict.decision_id,
        honored=True,
        action_taken="stopped",
    )
    # ...then fall back, notify a human, or pick a compliant alternative.

# Prefer to branch yourself? check() returns a Verdict and never raises on a deny.
verdict = client.controls.check("run_command", arguments={"cmd": cmd})
# verdict.allowed (bool) | verdict.reason | verdict.action | verdict.decision_id

if verdict.allowed:
    run_command(cmd)                     # allowed -> proceed
else:
    # The policy said STOP. Your code decides what happens -- and reports it honestly:
    #
    #   A) HONOR the deny (recommended): stop, report honored=True.
    client.controls.report(verdict.decision_id, honored=True, action_taken="stopped")
    #
    #   B) OVERRIDE the deny (your call): proceed anyway, report honored=False.
    #      run_command(cmd)
    #      client.controls.report(verdict.decision_id, honored=False, action_taken="proceeded")
    #
    # Honored vs overridden is exactly what the Adherence metric counts.

# Create a task
task = client.tasks.create(
    title="Fix authentication bug",
    description="Login fails for SSO users",
    priority="high",
    agent_name="dobby-backend-agent",
)
print(f"Task created: {task['id']}")

# List pending tasks
pending = client.tasks.list(status="pending")
for t in pending.get("tasks", []):
    print(f"  - {t['title']} ({t['status']})")

# Update task status
client.tasks.update(task["id"], status="in_progress")

# List pending approvals
pending = client.approvals.list(status="pending")

# Approve
client.approvals.approve("task_abc123", comment="LGTM, ship it!")

# Reject with reason
client.approvals.reject("task_def456", reason="Needs more test coverage")

# List fleet agents
agents = client.agents.list()

# Register an external agent
agent = client.agents.register(
    display_name="Research Agent",
    framework="crewai",
    protocol="a2a",
    endpoint_url="https://my-agent.example.com",
    capabilities=["research", "summarization"],
)
print(f"Agent registered: {agent['agent_id']}")
print(f"Gateway key: {agent['gateway_key_secret']}")

# Pause/resume
client.agents.pause(agent["agent_id"])
client.agents.resume(agent["agent_id"])

# Trigger an external agent immediately
result = client.agents.trigger("ext_abc123", payload={"topic": "AI governance"})
print(f"Trigger: {result['trigger_id']} — {result['status']}")

# Create a recurring schedule
client.agents.create_schedule("ext_abc123",
    name="Daily Blog Writer",
    schedule_config={"frequency": "daily", "time": "09:00"},
    trigger_payload={"topic": "AI governance"},
)

# Get trigger history with stats
history = client.agents.triggers("ext_abc123", include_stats=True)
print(f"Completed: {history['stats']['completed']}")

# Retry a failed trigger
client.agents.retry_trigger("ext_abc123", "trig_failed_id")

# Manage schedule
client.agents.update_schedule("ext_abc123", paused=True)
client.agents.delete_schedule("ext_abc123")

# Organization cost summary
summary = client.costs.summary(period="30d")
print(f"Total cost: ${summary.get('total_cost_usd', 0):.2f}")

# Per-agent breakdown
agents = client.costs.by_agent(period="7d")
for a in agents.get("agents", []):
    print(f"  {a['agent_name']}: ${a['cost_usd']:.4f}")

# Single agent deep-dive
profile = client.costs.agent_profile("ext_abc123")

from dobby_sdk import (
    DobbyAuthError,
    DobbyRateLimitError,
    DobbyBudgetExceededError,
    DobbyNotFoundError,
)

try:
    response = client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": "Hello"}],
    )
except DobbyAuthError:
    print("Invalid or expired API key")
except DobbyRateLimitError as e:
    print(f"Rate limited. Retry after: {e.retry_after}s")
except DobbyBudgetExceededError:
    print("Budget limit reached — contact your admin")
except DobbyNotFoundError:
    print("Resource not found")

from dobby_sdk import AsyncDobbyClient
import asyncio

async def main():
    async with AsyncDobbyClient(api_key="gk_user_...") as client:
        # Async chat completion
        response = await client.chat.completions.create(
            model="claude-sonnet-4-6",
            messages=[{"role": "user", "content": "Hello async!"}],
        )
        print(response.choices[0].message.content)

        # Async task creation
        task = await client.tasks.create(title="Async task")
        print(f"Created: {task['id']}")

asyncio.run(main())

# Environment variables (alternative to constructor params)
export DOBBY_API_KEY="gk_user_..."
export DOBBY_BASE_URL="https://dobby-ai.com"
export DOBBY_ORG_ID="org_..."
export DOBBY_TENANT_ID="tenant_..."

# Then just:
client = DobbyClient()  # Auto-reads from env