AgentRun Python SDK

Create agents, run them on your inputs, and collect the results. Sync and async clients, typed responses, Python 3.10 or newer.

pip install parcha-agentrun

The package on PyPI is parcha-agentrun. The module you import is agentrun.

Quick start

Set AGENTRUN_API_KEY in your environment, then:

from agentrun import Agent

agent = Agent(
    name="Company verifier",
    instructions="Verify the company and cite primary sources.",
)
run = agent.run("Verify acme.example")
result = agent.wait(run.id)

print(result.status)     # "completed", "failed", "cancelled", or "blocked"
print(result.markdown)   # the report, or None

Agent saves itself on the first call that needs it. Save agent.id and pass Agent(agent_id=...) next time to reuse it. Two things to know about wait. It returns for every terminal status, so check result.status before using the report. It gives up after 10 minutes by default and raises PollTimeoutError, but the run keeps going on the server. Call agent.get_run(run.id) later to fetch it.

Build an agent from a description

agent = Agent.create("Research companies and cite sources.")
# Save agent.build_id if this process might stop.
agent.wait_until_ready()   # builds can take a few minutes
run = agent.run("Research acme.example")

Resume an unfinished build with Agent(build_id=...). Any run method waits for the build first. A build that fails raises BuildError; read error.build.error for the reason.

Async

Every class has an async twin with the same methods. Works under asyncio and trio.

from agentrun import AsyncAgent

async def main() -> None:
    agent = AsyncAgent(agent_id="agent-123")
    run = await agent.run("Verify acme.example")
    result = await agent.wait(run.id)

Statuses

The API reports progress in its own vocabulary. The SDK maps it to a short list so is_done and your code stay simple. The original value is always in raw_status.

Batches use queued, running, completed, completed_with_errors, failed, and cancelled. Batch rows use queued, running, completed, and failed. A status the SDK does not recognise passes through unchanged.

Structured output

Pass a pydantic model or a JSON schema dict as output_schema. The result's structured_output is a dict; output_as validates it into your model.

from pydantic import BaseModel

class Finding(BaseModel):
    verified: bool
    evidence: str

run = agent.run("Verify acme.example", output_schema=Finding)
finding = agent.wait(run.id).output_as(Finding)

Batches

Run up to 100 inputs at once. The SDK checks the limits before sending anything.

batch = agent.batch(
    ["Verify acme.example", "Verify beta.example"],
    name="Q3 checks",
    max_concurrent=5,
    idempotency_key="q3-checks-001",
)
done = agent.wait_batch(batch.id)
for row in agent.get_batch_results(batch.id).results:
    print(row.row_number, row.status, row.run_id, row.headline)

Result rows carry headline, excerpt, report_link, and structured_output. Each row's input is on the batch itself: agent.get_batch(batch.id).rows gives row.question and row.row_data.

A CSV works too. Name the column that holds each input. The other columns come back as row.row_data on agent.get_batch(batch.id).rows.

from agentrun import BatchCSV

batch = agent.batch(BatchCSV("companies.csv", input_column="website"))

BatchCSV accepts a path, CSV text, bytes, or an open file. Files are limited to 1 MB.

wait_batch returns when the batch completes, fails, is cancelled, or pauses for attention. Check batch.paused_reason. get_batch(id, include_rows=False) skips per-row data on large batches. cancel_batch stops new rows; rows already running finish. retry_batch re-runs failed rows and spends credits.

Loops

A loop is a batch that repeats on a schedule. The first run starts immediately.

loop = agent.loop(
    "Check acme.example for new filings",
    {"frequency": "weekly", "schedule_config": {"hour": 9, "minute": 0, "day_of_week": 0}, "timezone": "America/New_York"},
    name="Weekly filings",
    max_credits=20,
)
history = agent.get_loop(loop.id)
agent.pause_loop(loop.id)
agent.delete_loop(loop.id, cancel_active=True)

frequency is daily, weekly, or monthly. Weekly uses day_of_week with 0 as Monday. Monthly uses day_of_month from 1 to 28. max_credits caps each occurrence, not the lifetime total. Inputs are frozen at creation; update_loop changes only the name and cadence.

Campaigns

A campaign is a multi-stage plan. Save it as a draft, review it, then approve it with the digest of the plan you reviewed.

draft = agent.create_campaign(plan)
# read draft.stages and draft.deliverables
started = agent.approve_campaign(draft.id, draft.review_digest)
final = agent.wait_campaign(started.id)
artifact = agent.download_campaign_artifact(final.id, "companies")
open(artifact.filename or "companies.xlsx", "wb").write(artifact.content)

An agent configured with run_mode="campaign" and a campaign_plan can start its template directly with agent.campaign({"goal": "..."}). That call starts work immediately. Approving with a stale digest raises ConflictError.

Skills and integrations

from agentrun import AgentRun

client = AgentRun()
skills = client.list_skills(query="company")
tools = client.list_integrations(query="company")

agent = client.agent(
    name="Company verifier",
    instructions="Verify the company and cite evidence.",
    skills=[s.name for s in skills.skills],
    tools=[t.name for t in tools.tools if not t.always_on],
)

Save your own Markdown as a skill with client.create_skill(name, description, content). Creating a skill with an existing name updates it; the response has created=False.

Errors

Every error inherits from AgentRunError.

APIError subclasses carry status_code, body, and request_id. Include the request ID when you report a problem. The SDK checks the limits the API enforces before sending anything, such as skill content of at least 50 characters or max_cost_usd between 0.5 and 1000, and raises ValueError locally. The SDK never retries a request on its own. Pass the same idempotency_key to run, batch, loop, or create_campaign when you retry, so the server does not start the work twice.

Configuration

base_url must use HTTPS. Plain HTTP is allowed only for localhost. Redirects are never followed. Use with AgentRun() as client: or async with AsyncAgentRun() as client: to close the connection pool.

Types

The package ships py.typed. Responses are pydantic models, so run.status and batch.rows have the types your editor shows. Records keep any field the server adds later, reachable as an attribute or with record["field"]. id is always the record's ID, whatever the API calls it. Agent records read back instructions, skills, and tools, the names Agent(...) takes; the API's system_prompt, skill_names, and mcp_tool_names are there too.

Development

uv sync --group dev
uv run pytest
uv run ruff check src tests
uv run pyright
uv run mypy
uv build && uv run twine check dist/*

Tests run against a fake API. Every test runs twice, once for the sync client and once for the async client.

Two tests tie the SDK to the backend:

• tests/test_contract.py validates every request the SDK sends, and every canned response, against tests/fixtures/public-v2.json, the backend's generated OpenAPI document. Refresh it with python scripts/update_contract.py <path-or-url> when the API changes.
• tests/test_packaged_contract.py starts a real local HTTP server and runs the same script the backend uses to verify packaged SDKs.

scripts/smoke_staging.py exercises a live deployment. It is read-only unless you pass --run.