> For the complete documentation index, see [llms.txt](https://docs.theacompute.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.theacompute.com/integrations/python.md). # Python In Python, I'm just the OpenAI SDK pointed at a different base URL. Swap one line and keep the rest of your code. When you need something the OpenAI client can't give you (my on-chain receipt headers, or async streaming you build by hand), reach for `httpx`. You'll need an API key, which you make in Settings. I only show it once. *** ## Using the OpenAI Python SDK ```bash pip install openai ``` ```python import os from openai import OpenAI client = OpenAI( base_url="https://api.theacompute.com/v1", api_key=os.environ["THEACOMPUTE_API_KEY"] ) # One answer, all at once response = client.chat.completions.create( model="qwen3-8b", messages=[ {"role": "system", "content": "You are a precise technical assistant."}, {"role": "user", "content": "Explain how the EVM executes smart contract bytecode."} ], max_tokens=512, temperature=0.3 ) print(response.choices[0].message.content) print(f"Tokens used: {response.usage.total_tokens}") ``` ### Let me stream it ```python stream = client.chat.completions.create( model="llama-3.3-70b", messages=[{"role": "user", "content": "Write a Python function to parse JSON safely."}], stream=True ) for chunk in stream: delta = chunk.choices[0].delta.content if delta: print(delta, end="", flush=True) print() ``` ### Going async ```python import asyncio from openai import AsyncOpenAI async def main(): client = AsyncOpenAI( base_url="https://api.theacompute.com/v1", api_key=os.environ["THEACOMPUTE_API_KEY"] ) async with client.chat.completions.stream( model="qwen3-8b", messages=[{"role": "user", "content": "What is a Merkle tree?"}] ) as stream: async for chunk in stream: delta = chunk.choices[0].delta.content if delta: print(delta, end="", flush=True) asyncio.run(main()) ``` *** ## Reading my on-chain headers with httpx I put the job's on-chain transactions in the headers of every response: `x-theacompute-tx-hash` is the escrow lock, and `x-theacompute-settlement-tx` is the settlement proving the job ran and got paid. `x-theacompute-credits-remaining` is your remaining units (my API calls units "credits"). The OpenAI SDK hides headers from you, so call my endpoint with `httpx` and read them from the raw response. ```python import os import httpx API_KEY = os.environ["THEACOMPUTE_API_KEY"] BASE_URL = "https://api.theacompute.com/v1" def chat(messages: list, model: str = "qwen3-8b") -> dict: response = httpx.post( f"{BASE_URL}/chat/completions", headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" }, json={"model": model, "messages": messages}, timeout=120 ) response.raise_for_status() data = response.json() data["_theacompute"] = { "job_id": response.headers.get("x-theacompute-job-id"), "escrow_tx": response.headers.get("x-theacompute-tx-hash"), "settlement_tx": response.headers.get("x-theacompute-settlement-tx"), "worker": response.headers.get("x-theacompute-worker"), "credits_remaining": response.headers.get("x-theacompute-credits-remaining") } return data result = chat([{"role": "user", "content": "What is ERC-4337 account abstraction?"}]) content = result["choices"][0]["message"]["content"] job_id = result["_theacompute"]["job_id"] settlement = result["_theacompute"]["settlement_tx"] print(content) print(f"\nVerify payment: https://robinhoodchain.blockscout.com/tx/{settlement}") ``` *** ## Streaming from me with plain httpx ```python import json import httpx def stream_chat(messages: list, model: str = "qwen3-8b"): with httpx.stream( "POST", f"{BASE_URL}/chat/completions", headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json" }, json={"model": model, "messages": messages, "stream": True}, timeout=120 ) as response: response.raise_for_status() for line in response.iter_lines(): if not line or line == "data: [DONE]": continue if line.startswith("data: "): chunk = json.loads(line[6:]) delta = chunk["choices"][0]["delta"].get("content", "") if delta: print(delta, end="", flush=True) stream_chat([{"role": "user", "content": "What is WebGPU?"}]) ``` *** ## Chatting with me over several turns ```python from openai import OpenAI client = OpenAI( base_url="https://api.theacompute.com/v1", api_key=os.environ["THEACOMPUTE_API_KEY"] ) messages = [{"role": "system", "content": "You are a knowledgeable assistant."}] while True: user_input = input("You: ").strip() if not user_input: break messages.append({"role": "user", "content": user_input}) response = client.chat.completions.create( model="qwen3-8b", messages=messages ) assistant_message = response.choices[0].message.content messages.append({"role": "assistant", "content": assistant_message}) print(f"Assistant: {assistant_message}") ``` *** ## When I send back an error A `402` means you're out of units, so top up in Settings with **Add more units**. A `503` means I'm out of workers for that model for now, and a `504` means the job timed out. Here's one way to retry sensibly: ```python from openai import OpenAI, APIStatusError, APIConnectionError import time client = OpenAI( base_url="https://api.theacompute.com/v1", api_key=os.environ["THEACOMPUTE_API_KEY"] ) def chat_with_retry(messages: list, model: str, max_retries: int = 3) -> str: for attempt in range(max_retries): try: response = client.chat.completions.create( model=model, messages=messages ) return response.choices[0].message.content except APIStatusError as e: if e.status_code == 402: raise Exception("Insufficient credits. Top up at theacompute.com/app") from e if e.status_code == 503: retry_after = e.response.json().get("error", {}).get("details", {}).get("retry_after", 10) if attempt < max_retries - 1: time.sleep(retry_after) continue raise if e.status_code == 504: # The job timed out, and I already refunded the units if attempt < max_retries - 1: time.sleep(5) continue raise raise except APIConnectionError as e: if attempt < max_retries - 1: time.sleep(2 ** attempt) continue raise raise Exception(f"Failed after {max_retries} attempts") ``` *** ## Asking me for your units balance `credits_remaining` is your units balance, and `usdg_value` is what those units are worth in USDG. ```python import httpx def get_balance() -> dict: response = httpx.get( "https://api.theacompute.com/v1/account", headers={"Authorization": f"Bearer {os.environ['THEACOMPUTE_API_KEY']}"} ) response.raise_for_status() return response.json() balance = get_balance() print(f"Credits remaining: {balance['credits_remaining']} (${balance['usdg_value']:.2f})") ``` --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.theacompute.com/integrations/python.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.