Metadata-Version: 2.4
Name: amber-sdk
Version: 0.1.2
Summary: Python SDK for defining durable Amber AI agent workflows
Project-URL: Repository, https://github.com/amber-runtime/amber
Author: Amber
License-Expression: MIT
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Requires-Dist: dbos-openai-agents>=0.3.0
Requires-Dist: dbos==2.20.0
Requires-Dist: psycopg2-binary>=2.9.12
Description-Content-Type: text/markdown

# Amber SDK

`amber-sdk` is the Python library for defining durable Amber agent workflows in
your agent applications. It provides the runtime object used by API and worker
processes, plus decorators for durable workflows and steps.

## Install

If you are deploying with Amber, install the full product package:

```bash
pip install amber-runtime
```

`amber-runtime` includes the `amber` CLI and depends on `amber-sdk`, so
application code can use `from amber import ...`.

Install `amber-sdk` directly only when you need the Python library without the
CLI:

```bash
pip install amber-sdk
```

## DB URL

Durable execution requires a Postgres database. Amber uses this database to store
workflow state so queued runs, steps, and sleeps can recover cleanly after
restarts. Set `DB_URL` in both the API and worker environments.

## Public API

Most agent applications use these imports from the `amber` module:

```python
from amber import (
    AgentRuntime,
    agent_runner,
    register_agent,
    sleep,
    step,
)
```

`AgentRuntime` is the recommended entry point for agent applications. The API
process uses it to enqueue agent runs, and the worker process uses it to execute
those queued runs.

`agent_runner` runs OpenAI Agents SDK agents inside Amber workflows. Use it
inside registered agents so agent calls are tracked as part of the durable
workflow and can recover cleanly after restarts.

`@register_agent` creates a named durable agent workflow that
`AgentRuntime.agents.start(...)` can enqueue. `@step` marks retryable units inside workflows and `sleep`
provides durable sleep that recover cleanly after restarts.

## Application Shape

Amber applications define a normal Python app and an agent runtime target, then
deploy with the `amber` CLI.

This example uses FastAPI for the app server and the OpenAI Agents SDK hosted
web search tool; install FastAPI separately if your app uses it.

```python
from agents import Agent, WebSearchTool
from fastapi import FastAPI

from amber import AgentRuntime, agent_runner, register_agent


research_agent = Agent(
    name="research-assistant",
    instructions="""You are a research assistant. Given a topic:
1. Search for current information when needed
2. Evaluate whether you have enough to write a thorough summary
3. If not, search again with a more specific or different query
4. Synthesize findings into a clear, well-structured summary
Be explicit about what you found and what remains uncertain.""",
    tools=[WebSearchTool()],
)


@register_agent(name="research-assistant")
async def research(topic: str) -> str:
    result = await agent_runner(
        starting_agent=research_agent,
        input=f"Research this topic thoroughly: {topic}",
    )
    return str(result.final_output)

agent_runtime = AgentRuntime(
    queue_name="agent-runs",
)

app = FastAPI(lifespan=agent_runtime.api_lifespan())

@app.post("/runs")
async def start_run(payload: dict[str, str]) -> dict[str, str]:
    handle = await agent_runtime.agents.start(
        "research-assistant",
        payload["input"],
    )
    return {"workflow_id": handle.workflow_id}
```

If agents are defined in the same module as `agent_runtime`, no `agent_modules`
setting is needed. If agents live in separate files, import those modules in the
backend module so `@register_agent` runs before API requests enqueue work, and
list the same modules in `AgentRuntime(agent_modules=[...])` so the worker
imports them when it starts.

```python
# my_app/agents.py
from amber import register_agent


@register_agent(name="research-assistant")
async def research(topic: str) -> str:
    ...
```

```python
# my_app/main.py
from fastapi import FastAPI
from amber import AgentRuntime
from . import agents  # registers @register_agent workflows

agent_runtime = AgentRuntime(
    agent_modules=["my_app.agents"],
    queue_name="agent-runs",
)

app = FastAPI(lifespan=agent_runtime.api_lifespan())

@app.post("/runs")
async def start_run(payload: dict[str, str]) -> dict[str, str]:
    handle = await agent_runtime.agents.start(
        "research-assistant",
        payload["input"],
    )
    return {"workflow_id": handle.workflow_id}
```

Run the API process with your ASGI server:

```bash
uvicorn my_app.main:app
```

Run a worker process against the same `AgentRuntime` target:

```bash
python -m amber.worker my_app.main:agent_runtime
```

## Worker Concurrency

`worker_concurrency` defaults to `8`. To change how many workflows each worker
process can run at once, set `worker_concurrency=<number>` on `AgentRuntime`.

`queue_concurrency` defaults to `None`, meaning there is no global cap for the
queue. Set `queue_concurrency=<number>` only when you need a maximum concurrency
across workers.

The number of worker processes is controlled by however you run or deploy
workers, not by SDK queue settings.

```python
agent_runtime = AgentRuntime(
    queue_name="agent-runs",
    worker_concurrency=16,
    queue_concurrency=64,
)
```

## Deploying

Use `amber-runtime` for the end-to-end product workflow:

```bash
pip install amber-runtime
amber init
amber deploy
```

See the `amber-runtime` package documentation for deployment, dashboard access,
and workflow visibility.
