How to Call an AI Agent API from Python

Overview

Every Runcept agent is a standard HTTP API. From Python, you start a job, poll until it's complete, and consume the typed JSON output.

Prerequisites

With httpx (async)

import asyncio
import httpx
import os

RUNCEPT_API_KEY = os.environ["RUNCEPT_API_KEY"]
BASE = "https://www.runcept.com/api/v1"

async def run_agent(agent_slug: str, input_data: dict) -> dict:
    headers = {
        "Authorization": f"Bearer {RUNCEPT_API_KEY}",
        "Content-Type": "application/json",
    }
    async with httpx.AsyncClient(timeout=10) as client:
        # Start the job
        start = await client.post(
            f"{BASE}/run",
            headers=headers,
            json={"agent": agent_slug, "input": input_data},
        )
        start.raise_for_status()
        job_id = start.json()["job_id"]

        # Poll for result
        while True:
            await asyncio.sleep(2)
            result = await client.get(f"{BASE}/jobs/{job_id}", headers=headers)
            result.raise_for_status()
            data = result.json()
            if data["status"] == "complete":
                return data["output"]
            if data["status"] == "failed":
                raise RuntimeError(f"Agent failed: {data.get('error')}")

async def main():
    output = await run_agent("runcept-echo", {"text": "hello from python"})
    print(output)

asyncio.run(main())

With requests (sync)

import time
import requests
import os

def run_agent(agent_slug: str, input_data: dict) -> dict:
    headers = {"Authorization": f"Bearer {os.environ['RUNCEPT_API_KEY']}"}
    r = requests.post(
        "https://www.runcept.com/api/v1/run",
        headers=headers,
        json={"agent": agent_slug, "input": input_data},
    )
    r.raise_for_status()
    job_id = r.json()["job_id"]

    while True:
        time.sleep(2)
        poll = requests.get(
            f"https://www.runcept.com/api/v1/jobs/{job_id}",
            headers=headers,
        )
        poll.raise_for_status()
        data = poll.json()
        if data["status"] == "complete":
            return data["output"]
        if data["status"] == "failed":
            raise RuntimeError(data.get("error"))

result = run_agent("pr-description", {"diff": open("my.diff").read()})
print(result["description"])

Rate Limiting

import time
from requests.exceptions import HTTPError

def run_agent_with_retry(slug, input_data, max_retries=3):
    for attempt in range(max_retries):
        try:
            return run_agent(slug, input_data)
        except HTTPError as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 60))
                time.sleep(retry_after)
            else:
                raise
    raise RuntimeError("Max retries exceeded")

Next Steps

Related guides