Skip to main content
Version: 0.8

Python REST API Quickstart

Query Nexatron from Python using the REST API directly with httpx or requests.

Authentication​

All API requests require an API key passed in the Authorization header:

Authorization: Bearer nxa_your_api_key_here

Generate an API key from Settings > API Keys in the Nexatron dashboard.

import httpx

BASE_URL = "https://api.nexatron.dev"
API_KEY = "nxa_your_api_key_here"

client = httpx.Client(
    base_url=BASE_URL,
    headers={"Authorization": f"Bearer {API_KEY}"},
    timeout=60.0,
)

Execute a Query​

response = client.post("/api/v1/query", json={
    "question": "What were total sales last quarter?",
})
response.raise_for_status()

result = response.json()

print(result["sql"])
# SELECT SUM(amount) AS total_sales
# FROM orders
# WHERE order_date >= '2026-01-01' AND order_date < '2026-04-01'

print(result["data"])
# [{"total_sales": 1284350.00}]

print(result["summary"])
# "Total sales last quarter were $1,284,350.00."

Query with Conversation Context​

Pass a conversation_id to maintain context across follow-up questions:

# First question
r1 = client.post("/api/v1/query", json={
    "question": "Show revenue by region",
})
conversation_id = r1.json()["conversation_id"]

# Follow-up question (references previous results)
r2 = client.post("/api/v1/query", json={
    "question": "Which region grew the fastest?",
    "conversation_id": conversation_id,
})
print(r2.json()["summary"])

Stream Results​

Use server-sent events for real-time streaming:

import json

with client.stream(
    "POST",
    "/api/v1/query/stream",
    json={"question": "Monthly revenue for 2025"},
) as response:
    for line in response.iter_lines():
        if not line or not line.startswith("data: "):
            continue

        payload = json.loads(line[6:])
        event_type = payload.get("type")

        if event_type == "stage":
            print(f"  [{payload['stage']}] {payload['message']}")
        elif event_type == "sql":
            print(f"  SQL: {payload['sql']}")
        elif event_type == "data":
            print(f"  Rows: {len(payload['rows'])}")
        elif event_type == "summary":
            print(f"  Summary: {payload['text']}")
        elif event_type == "done":
            print(f"  Conversation: {payload['conversation_id']}")
        elif event_type == "error":
            print(f"  Error: {payload['message']}")

Using requests​

import requests

BASE_URL = "https://api.nexatron.dev"
headers = {"Authorization": "Bearer nxa_your_api_key_here"}

# Execute a query
response = requests.post(
    f"{BASE_URL}/api/v1/query",
    headers=headers,
    json={"question": "Top 10 customers by lifetime value"},
    timeout=60,
)
response.raise_for_status()

result = response.json()
for row in result["data"]:
    print(f"  {row['customer_name']}: ${row['lifetime_value']:,.2f}")

Export Results​

Export as CSV​

response = client.post("/api/v1/query", json={
    "question": "All orders from last month",
    "format": "csv",
})

with open("orders.csv", "w") as f:
    f.write(response.text)

Export as JSON​

response = client.post("/api/v1/query", json={
    "question": "All orders from last month",
})

import json
with open("orders.json", "w") as f:
    json.dump(response.json()["data"], f, indent=2)

Load into pandas​

import pandas as pd

response = client.post("/api/v1/query", json={
    "question": "Daily active users for the past 90 days",
})
result = response.json()

df = pd.DataFrame(result["data"])
print(df.describe())

Async Usage​

import httpx
import asyncio

async def main():
    async with httpx.AsyncClient(
        base_url="https://api.nexatron.dev",
        headers={"Authorization": "Bearer nxa_your_api_key_here"},
        timeout=60.0,
    ) as client:
        response = await client.post("/api/v1/query", json={
            "question": "Revenue by product category",
        })
        result = response.json()
        print(result["summary"])

asyncio.run(main())

Error Handling​

try:
    response = client.post("/api/v1/query", json={
        "question": "Show me the data",
    })
    response.raise_for_status()
except httpx.HTTPStatusError as e:
    if e.response.status_code == 401:
        print("Invalid or expired API key")
    elif e.response.status_code == 422:
        error = e.response.json()
        print(f"Validation error: {error['detail']}")
    elif e.response.status_code == 429:
        print("Rate limited. Retry after:", e.response.headers.get("Retry-After"))
    else:
        print(f"API error {e.response.status_code}: {e.response.text}")
except httpx.TimeoutException:
    print("Request timed out. Try a simpler query or increase the timeout.")

Next Steps​