Skip to Content
SdkWorking with Agents

Working with Agents

The Curate-Me SDK manages agents through the client.agents sub-client. Agents are server-side resources identified by an opaque agent_id. You create, list, update, and run them through the SDK — the platform handles execution, governance, and cost recording for you. There is no local base class to subclass.

from curate_me import CurateMe
 
# Initialize the client (or use CurateMe.from_env())
client = CurateMe(api_key="cm_sk_xxx", org_id="org_xxx")

Listing and Fetching Agents

# List agents (async)
agents = await client.agents.list(limit=50)
for agent in agents:
    print(agent.id, agent.name, agent.model, agent.status)
 
# Fetch a single agent by ID
agent = await client.agents.get("agt_123")

Every method has a synchronous variant suffixed with _sync:

agents = client.agents.list_sync(limit=50)
agent = client.agents.get_sync("agt_123")

Creating an Agent

Provide a name and a model identifier. description, config, and metadata are optional.

agent = await client.agents.create(
    name="content-analysis",
    model="claude-sonnet-4-6",
    description="Analyzes content and extracts structured metadata.",
    config={
        "temperature": 0.1,
        "max_tokens": 1024,
    },
)
print(agent.id)  # e.g. "agt_123"

Running an Agent

Call client.agents.run() with the agent_id and an input dict. The call returns an AgentRun describing the completed (or failed) execution.

run = await client.agents.run(
    "agt_123",
    input={"content": "The product launch was a success. Sales exceeded expectations."},
)
 
print(run.status)        # ExecutionStatus, e.g. "completed"
print(run.output)        # the agent's result dict
print(run.cost)          # cost in USD for this run
print(run.tokens_used)   # total tokens consumed

run() accepts optional config, metadata, and timeout (seconds) arguments to override runtime behaviour for a single invocation.

The AgentRun Result

AgentRun carries the outcome of an execution:

FieldPurpose
idUnique run ID
agent_idThe agent that was run
statusExecutionStatus — pending, running, completed, failed, cancelled, or timeout
inputThe input data passed to the run
outputThe agent’s result dict (when completed)
errorError message (when failed)
duration_msWall-clock duration of the run
costCost of the run in USD
tokens_usedTotal tokens consumed
trace_idW3C trace ID for distributed tracing

Streaming Execution

For real-time progress, use client.agents.stream(). It yields StreamEvent objects as the run proceeds.

async for event in client.agents.stream(
    "agt_123",
    input={"content": "Long document to analyze..."},
):
    print(event.type, event.data)

Each StreamEvent has a type drawn from the EventType enum and a data payload:

Event TypePurpose
agent_startThe agent has begun execution
tokenStreams individual tokens during generation
agent_progressReports intermediate progress
agent_completeSuccessful completion with a result
agent_errorAn error occurred during execution
tool_call / tool_resultTool invocation and its result
guardrail_check / guardrail_violationGovernance guardrail events

Managing Runs

# List historical runs for an agent
runs = await client.agents.list_runs("agt_123", limit=20)
 
# Fetch a single run
run = await client.agents.get_run("agt_123", "run_456")
 
# Cancel an in-progress run
run = await client.agents.cancel_run("agt_123", "run_456")

Updating and Deleting

# Update fields on an agent
agent = await client.agents.update(
    "agt_123",
    description="Updated description",
    model="claude-opus-4-7",
)
 
# Delete an agent
await client.agents.delete("agt_123")

The @agent Decorator

If you want to track a local Python function as a Curate-Me agent — adding tracing and agent-run records — use the @agent decorator. The decorated function takes an input dict and returns a result dict.

from curate_me import agent
 
@agent(name="summarizer", model="claude-sonnet-4-6", tags=["analysis"])
async def summarize_text(input: dict) -> dict:
    text = input["text"]
    # ... your summarization logic ...
    return {"summary": "..."}
 
result = await summarize_text({"text": "Some long passage of text."})

For lighter-weight tracing of any function (not just agents), use the @trace decorator:

from curate_me import trace
 
@trace(name="my_operation")
async def my_function(query: str) -> str:
    return await process(query)

Governance Is Applied Server-Side

Note: Cost caps, PII scanning, rate limits, model allowlists, and HITL approvals are enforced by the Curate-Me gateway, not by the SDK. When you run an agent, the platform applies the governance chain and records cost automatically. You do not wire these checks into your own code. See the governance chain docs for the policy pipeline.

Testing

The SDK’s agent surface is exercised by the package test suite:

cd packages/sdk-python
poetry run pytest tests/ -v

To run the gateway/agent tests on the backend:

cd services/backend
poetry run pytest tests/services/autopilot/ -v

Defining Reusable Templates (Autopilot)

To ship a reusable, governed agent template that runs on the platform’s Autopilot engine (rather than a one-off run), author a template under services/backend/src/services/autopilot/templates/ with a SKILL.md contract. See the Autopilot template contract  and existing templates such as dev_team, cfo, and security_audit for reference.

SDK OverviewCLI Reference