camelai-run

The Python SDK for camelRun: durable agents you upsert by key and run, with tools that are ordinary functions in your code. The runtime runs the model loop, keeps each agent's history and files, and runs model-written code in a sandbox that can only call your tools.

pip install camelai-run

Python 3.11 or later. Create an API key at https://run.camelai.com/console/tokens and export it as CAMELAI_API_KEY.

Coding agents: this version's docs are in the package, next to its code (SKILL.md and sdk.md in camelai_run's directory: python -c "import camelai_run, os; print(os.path.dirname(camelai_run.__file__))"). To set camelRun up in a project, read https://run.camelai.com/SKILL.md.

The SDK is async-only: call it inside async def, and from synchronous code (a script, a Django view, a Celery task) with asyncio.run(...).

import asyncio
from camelai_run import Agents, tool

@tool
def weather(city: str) -> dict:
    """Today's weather in a city"""
    return {"city": city, "forecast": "sunny", "highC": 24}  # runs here, in your process

async def main():
    async with Agents() as agents:
        agent = await agents.upsert("quickstart", instructions="You are a concise assistant.", tools=[weather])
        run = await agent.run("Should I bring an umbrella in Lisbon today?")
        print(run.text)

asyncio.run(main())

• Keyed agents. upsert(key, ...) makes the agent for your key, or brings the existing one to the configuration given; its history and files last until you delete it.
• Runs. run() returns a Run (status, text, inputs, error, tool_errors, tool_calls) and raises RunError on failure (unless throw_on_error=False). No timeout unless you pass timeout=. agent.stream() yields text, tool calls and results as they happen, then the run.
• Tools. @tool takes async or plain functions (plain ones run in a thread), timeout= in seconds, and needs_approval=True. context.idempotency_key is stable across retries; context.progress("...") reports progress.
• Where tools run. Tools given to upsert run in that process, and one process at a time serves an agent's tools, only while it runs. With several processes (uvicorn or gunicorn workers, Celery, serverless) or deploys that restart them, serve tools over HTTP with serve_tools (below) instead.
• People in the loop. await run.inputs[0].answer(True, from_="alice") resumes a run waiting on approval.
• Events. on_event may be a plain or an async function; it runs in order, apart from the connection. close() stops it: events still queued are dropped.

Documentation: Quickstart, Concepts, SDK reference, and all of it as Markdown at https://run.camelai.com/llms.txt.

Serving tools to many users

When one server answers tools for many users' agents, serve them over HTTP and let the runtime say who each call is for. serve_tools is an ASGI app that verifies the runtime's signed identity token on every request and hands each call a context.identity:

pip install "camelai-run[server]"

from camelai_run import ToolContext, serve_tools, tool

@tool
async def list_todos(context: ToolContext) -> dict:
    """The current user's to-dos"""
    who = context.identity  # user (the actor, else the agent's subject), subject, tenant, agent, context
    return {"todos": await db.todos(user=who.user, team=who.context["team"])}

# tenant: yours (GET /v1/me): tokens for other tenants' agents, which may claim any user, are refused.
app = serve_tools([list_todos], runtime="https://run.camelai.com", tenant="acme")  # uvicorn, or mount in FastAPI

Name the server in a definition, make each user's agent from it, and run it as that user. Any process can do this (no tools here: the server above answers them), so every web worker and task can:

definition = await agents.runtime.upsert_definition("todos", name="Todos", mcpServers=[
    {"name": "todos", "url": "https://todos.example.com/mcp", "auth": {"type": "runtime"}}])
agent = await agents.upsert(f"todos-{user.id}", definition=definition["id"], subject=user.id, context={"team": user.team})
run = await agent.run("What's left for this week?", user=user.id)

Definitions take the REST API's field names (systemPrompt, mcpServers, openApi). The same @tool functions get the same identity when attached to an agent. verify_runtime_token(token, runtime=..., tenant=..., audience=...) checks a token on its own, and TestRuntime() signs tokens for tests: await TestRuntime().call_tool(app, url, "list_todos", {}, subject="alice").

Keep the API key on your backend: it can create and control every agent in your tenant. Sign in at https://run.camelai.com/console to add provider keys, create API tokens and watch agents. The TypeScript SDK is @camelai/run.

Asking the user

A tool marked @tool(needs_approval=True) is approved before each call; inside a tool, context.confirm(message), context.ask(message, schema) and context.require_url(url, message) ask the user. The run then returns with status == "input_required", and answering its inputs resumes it:

@tool
async def delete_app(app: str, context: ToolContext) -> dict:
    """Delete an app"""
    # Ask first: the call ends here, and runs again with the answer.
    if not await context.confirm(f"Delete {app}? Its URL stops working."):
        return {"cancelled": True}
    return await apps.delete(app, idempotency_key=context.idempotency_key)

agent = await agents.upsert("ops", tools=[delete_app])
run = await agent.run("Delete the demo app", user="alice")
while run.status == "input_required":
    run = await run.inputs[0].answer(True, from_="alice")

Everything in a tool before an ask runs again when the user answers. agent.pending_inputs() lists what an agent waits on, and agents.runtime.inbox(state="pending") what all your agents do.