Skip to main content

Python SDK

The official Python SDK for Deeployd provides full access to the platform API with type hints and async support.

Installation​

pip install deeployd
# or
poetry add deeployd

Quick Start​

from deeployd import Deeployd

client = Deeployd(api_key="your-api-key")

# Create an agent
agent = client.agents.create(
    name="My Assistant",
    system_prompt="You are a helpful assistant.",
    model="claude-sonnet-4-20250514"
)

# Start a conversation
conversation = client.conversations.create(agent_id=agent.id)

# Send a message
response = client.conversations.chat(
    conversation.id,
    message="Hello!"
)

print(response.content)

Async Support​

import asyncio
from deeployd import AsyncDeeployd

async def main():
    client = AsyncDeeployd(api_key="your-api-key")

    agent = await client.agents.create(
        name="My Assistant",
        system_prompt="You are helpful.",
        model="claude-sonnet-4-20250514"
    )

    conversation = await client.conversations.create(agent_id=agent.id)

    response = await client.conversations.chat(
        conversation.id,
        message="Hello!"
    )

    print(response.content)

asyncio.run(main())

Configuration​

Basic Configuration​

from deeployd import Deeployd

client = Deeployd(
    api_key="your-api-key",
    base_url="https://api.deeployd.com/v1",  # Optional
    timeout=30.0,  # Optional, in seconds
)

Environment Variables​

# .env
DEEPLOYD_API_KEY=your-api-key
DEEPLOYD_BASE_URL=https://api.deeployd.com/v1
# Auto-reads from environment
client = Deeployd()

Custom HTTP Client​

import httpx
from deeployd import Deeployd

# Custom httpx client
http_client = httpx.Client(
    headers={"X-Custom-Header": "value"},
    timeout=60.0
)

client = Deeployd(
    api_key="your-api-key",
    http_client=http_client
)

Agents API​

Create Agent​

agent = client.agents.create(
    name="Customer Support",
    description="Handles customer inquiries",
    system_prompt="You are a friendly customer support agent...",
    model="claude-sonnet-4-20250514",
    tools=["memory", "http_request", "calculator"],
    settings={
        "temperature": 0.7,
        "max_tokens": 4096
    }
)

Get Agent​

agent = client.agents.get("agent-123")

print(f"ID: {agent.id}")
print(f"Name: {agent.name}")
print(f"Status: {agent.status}")
print(f"Created: {agent.created_at}")

List Agents​

agents = client.agents.list(
    limit=20,
    offset=0,
    status="active"
)

for agent in agents.data:
    print(agent.name)

Update Agent​

client.agents.update(
    "agent-123",
    name="Updated Name",
    system_prompt="Updated prompt..."
)

Delete Agent​

client.agents.delete("agent-123")

Conversations API​

Create Conversation​

conversation = client.conversations.create(
    agent_id="agent-123",
    metadata={
        "user_id": "user-456",
        "session_id": "abc123"
    }
)

Send Message​

response = client.conversations.chat(
    conversation.id,
    message="What is the weather today?",
    attachments=[
        {
            "type": "file",
            "url": "https://example.com/document.pdf"
        }
    ]
)

print(response.content)
print(response.tool_calls)

Stream Response​

stream = client.conversations.chat_stream(
    conversation.id,
    message="Write a long story about..."
)

for chunk in stream:
    print(chunk.content, end="", flush=True)

Async Streaming​

async def stream_response():
    async for chunk in client.conversations.chat_stream(
        conversation.id,
        message="Write a story..."
    ):
        print(chunk.content, end="", flush=True)

Get Conversation History​

messages = client.conversations.get_messages(
    conversation.id,
    limit=50
)

for message in messages.data:
    print(f"{message.role}: {message.content}")

List Conversations​

conversations = client.conversations.list(
    agent_id="agent-123",
    limit=20
)

Delete Conversation​

client.conversations.delete(conversation.id)

Tools API​

List Available Tools​

tools = client.tools.list()

for tool in tools.data:
    print(f"Name: {tool.name}")
    print(f"Description: {tool.description}")
    print(f"Category: {tool.category}")

Create Custom Tool​

tool = client.tools.create(
    name="get_stock_price",
    description="Get the current stock price for a symbol",
    type="http",
    config={
        "method": "GET",
        "url": "https://api.stocks.com/v1/price/${symbol}",
        "headers": {
            "Authorization": "Bearer ${secrets.STOCKS_API_KEY}"
        }
    },
    parameters={
        "type": "object",
        "properties": {
            "symbol": {
                "type": "string",
                "description": "Stock symbol (e.g., AAPL)"
            }
        },
        "required": ["symbol"]
    }
)

Update Tool​

client.tools.update(
    "tool-123",
    description="Updated description"
)

Delete Tool​

client.tools.delete("tool-123")

Tasks API​

Create Task​

task = client.tasks.create(
    agent_id="agent-123",
    name="Generate Report",
    type="manual",
    input={
        "report_type": "weekly",
        "start_date": "2024-01-01"
    },
    priority="high"
)

Run Task​

execution = client.tasks.run(
    "task-123",
    input={"custom_param": "value"}
)

print(f"Execution ID: {execution.id}")
print(f"Status: {execution.status}")
print(f"Output: {execution.output}")

Get Task Status​

task = client.tasks.get("task-123")

print(f"Status: {task.status}")
print(f"Last Run: {task.last_run_at}")
print(f"Next Run: {task.next_run_at}")

List Tasks​

tasks = client.tasks.list(
    agent_id="agent-123",
    status="active"
)

Schedules API​

Create Schedule​

schedule = client.schedules.create(
    task_id="task-123",
    name="Daily Report",
    frequency="cron",
    cron_expression="0 9 * * *",  # 9 AM daily
    timezone="Europe/Paris",
    enabled=True
)

Update Schedule​

client.schedules.update(
    "schedule-123",
    cron_expression="0 10 * * *",  # Changed to 10 AM
    enabled=True
)

Pause/Resume Schedule​

# Pause
client.schedules.pause("schedule-123")

# Resume
client.schedules.resume("schedule-123")

Triggers API​

Create Webhook Trigger​

trigger = client.triggers.create(
    task_id="task-123",
    name="GitHub Push",
    type="webhook",
    config={
        "secret": "webhook-secret",
        "input_mapping": {
            "branch": "$.ref",
            "commit": "$.head_commit.id"
        }
    }
)

print(f"Webhook URL: {trigger.webhook_url}")
# https://api.deeployd.com/v1/triggers/trigger-123/webhook

List Triggers​

triggers = client.triggers.list(task_id="task-123")

Memory API​

Store Memory​

client.memory.store(
    "agent-123",
    key="user_preference",
    value={
        "theme": "dark",
        "language": "en"
    },
    ttl=86400  # Optional: expire after 24 hours
)

Retrieve Memory​

memory = client.memory.get("agent-123", "user_preference")
print(memory.value)

Search Memory​

results = client.memory.search(
    "agent-123",
    query="user preferences",
    limit=10
)

Delete Memory​

client.memory.delete("agent-123", "user_preference")

Teams API​

Create Team​

team = client.teams.create(
    name="Engineering",
    description="Engineering team"
)

Add Members​

client.teams.add_member(
    "team-123",
    user_id="user-456",
    role="member"
)

Remove Members​

client.teams.remove_member("team-123", "user-456")

Store API​

Browse Listings​

listings = client.store.browse(
    category="support",
    pricing="free",
    sort_by="popular",
    limit=20
)

Install Listing​

install = client.store.install("listing-123")
print(f"New agent ID: {install.agent_id}")

Create Review​

client.store.create_review(
    "listing-123",
    rating=5,
    title="Great agent!",
    content="This agent saved me hours of work."
)

Error Handling​

Exception Types​

from deeployd import (
    Deeployd,
    DeeploidError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError
)

try:
    client.agents.get("nonexistent")
except NotFoundError:
    print("Agent not found")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except ValidationError as e:
    print(f"Validation error: {e.errors}")
except DeeploidError as e:
    print(f"API error: {e.message}")

Retry Configuration​

client = Deeployd(
    api_key="your-api-key",
    max_retries=3,
    retry_delay=1.0,
    retry_max_delay=30.0,
    retry_on_status=[429, 500, 502, 503, 504]
)

Type Hints​

from deeployd import Deeployd
from deeployd.types import Agent, Conversation, Message, Tool

client = Deeployd()

agent: Agent = client.agents.get("agent-123")
conversation: Conversation = client.conversations.create(agent_id=agent.id)

Generic Responses​

from deeployd.types import PaginatedResponse, Agent

response: PaginatedResponse[Agent] = client.agents.list()

print(f"Data: {response.data}")
print(f"Total: {response.total}")
print(f"Limit: {response.limit}")
print(f"Offset: {response.offset}")

Pagination​

Manual Pagination​

offset = 0
limit = 20
has_more = True

while has_more:
    response = client.agents.list(limit=limit, offset=offset)

    for agent in response.data:
        print(agent.name)

    offset += limit
    has_more = len(response.data) == limit

Auto-Pagination​

for agent in client.agents.list_all():
    print(agent.name)

Async Auto-Pagination​

async for agent in client.agents.list_all():
    print(agent.name)

Webhooks​

Verify Webhook Signature​

from deeployd import verify_webhook_signature
from flask import Flask, request

app = Flask(__name__)

@app.post("/webhook")
def webhook():
    signature = request.headers.get("X-Deeployd-Signature")
    timestamp = request.headers.get("X-Deeployd-Timestamp")

    is_valid = verify_webhook_signature(
        payload=request.data,
        signature=signature,
        timestamp=timestamp,
        secret=os.environ["WEBHOOK_SECRET"]
    )

    if not is_valid:
        return "Invalid signature", 401

    # Process webhook...
    return "OK"

Context Managers​

Sync Client​

from deeployd import Deeployd

with Deeployd() as client:
    agent = client.agents.get("agent-123")
    print(agent.name)
# Client is automatically closed

Async Client​

from deeployd import AsyncDeeployd

async def main():
    async with AsyncDeeployd() as client:
        agent = await client.agents.get("agent-123")
        print(agent.name)

Logging​

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

# Or specifically for deeployd
logging.getLogger("deeployd").setLevel(logging.DEBUG)

Best Practices​

Use Environment Variables​

import os
from deeployd import Deeployd

# Don't hardcode API keys
client = Deeployd(api_key=os.environ.get("DEEPLOYD_API_KEY"))

Handle Errors Gracefully​

from deeployd import NotFoundError

def safe_get_agent(client, agent_id):
    try:
        return client.agents.get(agent_id)
    except NotFoundError:
        return None

Use Streaming for Long Responses​

# Better for long responses
for chunk in client.conversations.chat_stream(id, message=message):
    # Process incrementally
    print(chunk.content, end="")

Reuse Client Instance​

# Create once
client = Deeployd(api_key=api_key)

# Reuse across requests
def get_agent(agent_id):
    return client.agents.get(agent_id)

Use Context Managers​

# Ensures proper cleanup
with Deeployd() as client:
    # ... operations

Next: Learn about the CLI Reference for command-line usage.