Askpilot Public API Python Client SDK

The official Python client for the Askpilot public API. It gives you typed access to your organization and its workflows, with retries, timeouts, and clear exceptions built in.

• Python 3.9 or later
• A sync client and an async client
• Two runtime dependencies: httpx and pydantic

Install

pip install askpilot

Or, with uv:

uv add askpilot

Quickstart

import askpilot

client = askpilot.Askpilot(api_key="ask_...")

# The organization your key belongs to
organization = client.organization.get()
print(organization.name)

# One page of workflows
page = client.workflows.list(limit=10)
for workflow in page.data:
    print(workflow.id, workflow.name)

# Every workflow, page by page
for workflow in client.workflows.iterate():
    print(workflow.name)

# One workflow
workflow = client.workflows.get("018f2c1e-8a4b-7c3d-9e5f-1a2b3c4d5e6f")

# Start a workflow
start = client.workflows.start(workflow.id, context="New lead: John Smith, wants a 2-bed flat.")
print(start.session_id)

client.close()

You can also use the client as a context manager, so it closes itself:

with askpilot.Askpilot() as client:
    print(client.organization.get().name)

Your API key

You create an API key in Askpilot, under the organization settings. Askpilot shows the key once, and it starts with ask_. Keep it in a secret store, never in your code or your repository.

Pass the key to the client, or set the ASKPILOT_API_KEY environment variable and create the client with no arguments:

client = askpilot.Askpilot()  # reads ASKPILOT_API_KEY

The key never appears in a log line, in an exception message, or in repr(client).

Errors

Every exception the SDK raises is a subclass of askpilot.AskpilotError, so one except can catch them all. An error response from the API becomes the APIError subclass of its HTTP status, and the error code from the response stays on the exception as code:

import askpilot

try:
    start = client.workflows.start(workflow_id, context="New lead: John Smith.")
except askpilot.ConflictError as error:
    if error.code == "api_trigger_not_configured":
        print("Add a trigger with the source API to this workflow in Askpilot.")
    else:
        raise
except askpilot.RateLimitError as error:
    print(f"Too many requests. Wait {error.retry_after} seconds.")
except askpilot.APIError as error:
    print(f"{error.status_code} {error.code}: {error.message} (request id {error.request_id})")
except askpilot.APIConnectionError:
    print("Couldn't reach Askpilot. Check your network and try again.")

Every APIError has status_code, code, message, request_id, details, and body. Include request_id when you contact support. The error codes of each call are listed in docs/organization.md and docs/workflows.md.

Pagination

list() returns one page. page.data holds the items, and page.pagination.next_cursor is the cursor of the next page, or None on the last page:

cursor = None
while True:
    page = client.workflows.list(limit=100, cursor=cursor)
    for workflow in page.data:
        print(workflow.name)
    cursor = page.pagination.next_cursor
    if cursor is None:
        break

iterate() runs this loop for you and yields each item. Stop on next_cursor, not on the size of a page: a short page can still have a next page. The cursor is opaque. Send it back as it is.

Retries and the timeout

The client sends a GET again after a connection error, a timeout, or a 502, 503, or 504 response, and it sends any request again after a 429. It waits 0.5, 1, then 2 seconds between tries, or the Retry-After of a 429 (at most 60 seconds). A POST is never retried for another reason: a start that reached Askpilot may have started the workflow, and a second start would start it again.

client = askpilot.Askpilot(max_retries=0)  # no retries
client = askpilot.Askpilot(timeout=10.0)  # 10 seconds per request

The default timeout is 30 seconds, and it applies to each phase of a request. Pass an httpx.Timeout for a value per phase.

Logging

The SDK logs one line per request at DEBUG on the logger askpilot, with the method, the path, the status, the duration, the request id, and what's left of your rate limit. Before each retry it logs one line at WARNING with the cause and the wait. No line contains your key, a header value, or a body. Two lines turn the DEBUG line on:

import logging

logging.basicConfig()
logging.getLogger("askpilot").setLevel(logging.DEBUG)

The async client

AsyncAskpilot has the same methods as coroutines, and iterate() is an async iterator:

import asyncio

import askpilot


async def main() -> None:
    async with askpilot.AsyncAskpilot() as client:
        organization = await client.organization.get()
        print(organization.name)
        async for workflow in client.workflows.iterate():
            print(workflow.name)


asyncio.run(main())

Your own httpx client

Pass an httpx.Client (or an httpx.AsyncClient) as http_client for a proxy, a custom CA, or a test transport. You own it: close() doesn't close it. The SDK sets its own URL, headers, and timeout on every request, so the base URL and the headers of your client aren't used.

import httpx

import askpilot

http = httpx.Client(proxy="http://proxy.internal:3128", verify="/etc/ssl/company-ca.pem")
client = askpilot.Askpilot(http_client=http)

Reference

• docs/organization.md: client.organization
• docs/workflows.md: client.workflows
• The API reference, with a console to try calls: https://public-api.askpilot.com/docs

License

MIT. See LICENSE.