Python SDK Reference

# Install from PyPI
pip install dailyai-agents

# Install with async support
pip install dailyai-agents[async]

# Install a specific version
pip install dailyai-agents==1.4.0

from dailyai import AgentOS

# Option 1: Pass API key directly
client = AgentOS(api_key="sk-your-api-key")

# Option 2: Use environment variable (recommended)
# Set DAILYAI_API_KEY in your environment
client = AgentOS()  # Reads DAILYAI_API_KEY automatically

# Option 3: Use a JWT token (from /auth/login)
client = AgentOS(token="eyJhbGciOiJIUzI1NiIs...")

agents = client.agents.list(category="trading")
for agent in agents:
    print(f"{agent.id}: {agent.name} ({agent.status})")
    print(f"  Strategies: {', '.join(agent.strategies)}")

# Synchronous run
signal = client.agents.trading.run(asset="AAPL", strategy="momentum")
print(signal.direction)    # "long"
print(signal.confidence)   # 0.87
print(signal.entry_price)  # 198.45

# Async run (returns immediately)
job = client.agents.trading.run(
    asset="AAPL",
    strategy="momentum",
    async_run=True
)
print(job.run_id)  # "run_abc123"

# Poll for result
result = job.wait()  # Blocks until complete
print(result.direction)

from datetime import datetime, timedelta

# Get recent high-confidence signals
signals = client.signals.list(
    min_confidence=0.8,
    since=datetime.utcnow() - timedelta(hours=24),
    limit=10
)

for s in signals:
    print(f"{s.direction:5s} {s.asset:8s} conf={s.confidence:.0%}")

# Paginate through all signals
offset = 0
all_signals = []
while True:
    batch = client.signals.list(limit=100, offset=offset)
    if not batch:
        break
    all_signals.extend(batch)
    offset += 100

signal = client.signals.get("sig_abc123")
print(signal.metadata.reasoning)

weather = client.signals.weather(
    region="us",
    commodity="natural_gas",
    severity="high"
)
for w in weather:
    print(f"{w.commodity}: {w.expected_impact}")

import asyncio
from dailyai import AgentOS

async def main():
    client = AgentOS()

    async for event in client.dashboard.stream(
        channels=["signal.new", "agent.status"],
        assets=["AAPL", "TSLA"]
    ):
        print(f"[{event.type}] {event.data}")

asyncio.run(main())

# Create a webhook
webhook = client.webhooks.create(
    url="https://your-app.com/hooks/dailyai",
    events=["signal.created", "agent.error"],
    secret="whsec_your_secret"
)
print(f"Created: {webhook.id}")

# List all webhooks
for wh in client.webhooks.list():
    print(f"{wh.id}: {wh.url} ({wh.status})")

# Delete a webhook
client.webhooks.delete("whk_abc123")

from dailyai import AgentOS
from dailyai.exceptions import (
    DailyAIError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError,
)

client = AgentOS()

try:
    signal = client.agents.trading.run(asset="AAPL", strategy="momentum")
except AuthenticationError:
    print("Check your API key")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except NotFoundError:
    print("Agent or asset not found")
except ValidationError as e:
    print(f"Invalid parameters: {e.message}")
except DailyAIError as e:
    print(f"API error: {e.status_code} — {e.message}")

# Access rate limit headers from last request
info = client.rate_limit_info
print(f"Remaining: {info.remaining}")   # Requests remaining
print(f"Limit: {info.limit}")           # Total allowed
print(f"Reset: {info.reset}")           # UTC timestamp of reset

# Configure custom retry behavior
client = AgentOS(
    max_retries=5,      # Retry up to 5 times on 429/5xx
    timeout=60.0,       # Increase timeout for long-running agents
)

pip install dailyai-agents[async]

import asyncio
from dailyai import AsyncAgentOS

async def main():
    client = AsyncAgentOS()  # Reads DAILYAI_API_KEY

    # Run multiple agents concurrently
    signals = await asyncio.gather(
        client.agents.trading.run(asset="AAPL", strategy="momentum"),
        client.agents.trading.run(asset="MSFT", strategy="breakout"),
        client.agents.trading.run(asset="GOOGL", strategy="sentiment"),
    )

    for signal in signals:
        print(f"{signal.asset}: {signal.direction} ({signal.confidence:.0%})")

    # List signals (async)
    recent = await client.signals.list(min_confidence=0.8)
    print(f"Found {len(recent)} high-confidence signals")

    # Stream dashboard events
    async for event in client.dashboard.stream():
        print(event)

asyncio.run(main())

async with AsyncAgentOS() as client:
    signal = await client.agents.trading.run(
        asset="AAPL",
        strategy="momentum"
    )
    print(signal)
# Connection pool automatically closed