> For the complete documentation index, see [llms.txt](https://docs.robinmesh.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.robinmesh.com/integrations/python.md). # Python From Python, the simplest route to RobinMesh is the ordinary OpenAI SDK pointed at the RobinMesh base URL. When you want things the OpenAI client cannot give you, such as the on-chain receipt headers or hand-rolled async streaming, drop down to `httpx`. *** ## The OpenAI Python SDK ```bash pip install openai ``` ```python import os from openai import OpenAI client = OpenAI( base_url="https://api.robinmesh.com/v1", api_key=os.environ["ROBINMESH_API_KEY"] ) # Non-streaming 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}") ``` ### Streaming ```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() ``` ### Async ```python import asyncio from openai import AsyncOpenAI async def main(): client = AsyncOpenAI( base_url="https://api.robinmesh.com/v1", api_key=os.environ["ROBINMESH_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 the on-chain headers with httpx Settlement data rides in on response headers, and the OpenAI SDK never surfaces those. Call the endpoint with `httpx` yourself when you need the transaction hashes. ```python import os import httpx API_KEY = os.environ["ROBINMESH_API_KEY"] BASE_URL = "https://api.robinmesh.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["_robinmesh"] = { "job_id": response.headers.get("x-robinmesh-job-id"), "escrow_tx": response.headers.get("x-robinmesh-tx-hash"), "settlement_tx": response.headers.get("x-robinmesh-settlement-tx"), "worker": response.headers.get("x-robinmesh-worker"), "credits_remaining": response.headers.get("x-robinmesh-credits-remaining") } return data result = chat([{"role": "user", "content": "What is ERC-4337 account abstraction?"}]) content = result["choices"][0]["message"]["content"] job_id = result["_robinmesh"]["job_id"] settlement = result["_robinmesh"]["settlement_tx"] print(content) print(f"\nVerify payment: https://robinhoodchain.blockscout.com/tx/{settlement}") ``` *** ## Streaming over raw 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?"}]) ``` *** ## Holding a conversation across turns ```python from openai import OpenAI client = OpenAI( base_url="https://api.robinmesh.com/v1", api_key=os.environ["ROBINMESH_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}") ``` *** ## Handling errors ```python from openai import OpenAI, APIStatusError, APIConnectionError import time client = OpenAI( base_url="https://api.robinmesh.com/v1", api_key=os.environ["ROBINMESH_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 robinmesh.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; the credits were already refunded 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") ``` *** ## Querying your credit balance ```python import httpx def get_balance() -> dict: response = httpx.get( "https://api.robinmesh.com/v1/account", headers={"Authorization": f"Bearer {os.environ['ROBINMESH_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.robinmesh.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.