# Python

> A complete Automate HTTP Server client in Python — standard library only, with endpoint discovery, readiness gating, dataclass-typed results, cancellation, and an OpenAI-SDK variant.

Source: https://developers.cephable.com/docs/automate-http-server/samples/python

---
Python 3.9+. The main client uses only the standard library, so it runs in a bare virtualenv; an `httpx` variant and an OpenAI-SDK variant follow.

---

## The client

```python
# cephable.py
"""Client for the Cephable Automate HTTP Server (local, loopback-only)."""

from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Union

DEFAULT_PORT = 4317
PORT_ATTEMPTS = 12
READY_STATUSES = {"idle", "terminated"}


class CephableRequestError(RuntimeError):
    """The request never produced a run: 400, 401, 404, 409."""

    def __init__(self, message: str, status: int, error_type: Optional[str] = None) -> None:
        super().__init__(message)
        self.status = status
        self.error_type = error_type

    @property
    def is_busy(self) -> bool:
        return self.status == 409

    @property
    def is_unauthorized(self) -> bool:
        return self.status == 401


class CephableRunError(RuntimeError):
    """A run happened and did not complete. Carries the full record."""

    def __init__(self, result: "RunResult") -> None:
        detail = f" ({result.error_code})" if result.error_code else ""
        super().__init__(f"Automate run {result.status}{detail}")
        self.result = result


@dataclass
class RunResult:
    raw: Dict[str, Any]

    @property
    def status(self) -> str:
        return self.raw["status"]

    @property
    def completed(self) -> bool:
        return self.raw["status"] == "completed"

    @property
    def answer(self) -> str:
        return self.raw.get("answer", "")

    @property
    def final_answer(self) -> Optional[str]:
        """Only present when answerContract was set."""
        return self.raw.get("finalAnswer")

    @property
    def value(self) -> str:
        """The contract value when there is one, otherwise the whole answer."""
        return self.raw.get("finalAnswer") or self.raw.get("answer", "")

    @property
    def error_code(self) -> Optional[str]:
        return self.raw.get("errorCode")

    @property
    def steps(self) -> List[Dict[str, Any]]:
        return self.raw.get("steps") or []

    @property
    def failed_steps(self) -> List[Dict[str, Any]]:
        return [step for step in self.steps if step.get("status") == "failed"]

    @property
    def produced_files(self) -> List[str]:
        """Resolved absolute paths of files and folders the run created or edited."""
        return [step["producedFilePath"] for step in self.steps if step.get("producedFilePath")]

    @property
    def duration_ms(self) -> int:
        return self.raw["durationMs"]

    @property
    def usage(self) -> Optional[Dict[str, Any]]:
        return self.raw.get("usage")

    @property
    def awaiting_tool_results(self) -> bool:
        """The run is parked waiting for this process to execute the tools in `tool_calls`."""
        return self.raw["status"] == "awaiting_tool_results"

    @property
    def tool_calls(self) -> List[Dict[str, Any]]:
        return self.raw.get("toolCalls") or []


@dataclass
class CephableClient:
    endpoint: Optional[str] = None
    token: str = field(default_factory=lambda: os.environ.get("CEPHABLE_AUTOMATE_KEY", ""))

    def __post_init__(self) -> None:
        if not self.token:
            raise ValueError("Set CEPHABLE_AUTOMATE_KEY or pass token=")

    # ── transport ────────────────────────────────────────────────────────────────

    def resolve_endpoint(self) -> str:
        """Cephable moves to the next free port when 4317 is taken. Sweep the same window."""
        if self.endpoint:
            return self.endpoint
        for offset in range(PORT_ATTEMPTS):
            candidate = f"http://127.0.0.1:{DEFAULT_PORT + offset}"
            try:
                status, body = self._raw_request(candidate, "/health", None, timeout=3.0)
            except OSError:
                continue  # nothing listening here
            # A 401 still proves a server is listening — report the key, not the sweep.
            if status == 401:
                raise CephableRequestError(
                    f"Cephable is listening on {candidate} but rejected the access key",
                    401,
                    "authentication_error",
                )
            if status == 200 and body.get("service") == "cephable-agent":
                self.endpoint = candidate
                return candidate
        raise RuntimeError(
            f"No Cephable Automate server answered on 127.0.0.1:{DEFAULT_PORT}-"
            f"{DEFAULT_PORT + PORT_ATTEMPTS - 1}. Open Cephable and enable "
            "Extensions > Cephable features > Build & Extend > Automate HTTP Server."
        )

    def _raw_request(
        self,
        endpoint: str,
        route: str,
        payload: Optional[Dict[str, Any]],
        timeout: float,
    ) -> tuple[int, Dict[str, Any]]:
        data = json.dumps(payload).encode("utf-8") if payload is not None else None
        request = urllib.request.Request(
            f"{endpoint}{route}",
            data=data,
            method="POST" if data is not None else "GET",
            headers={
                "authorization": f"Bearer {self.token}",
                "content-type": "application/json",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return response.status, json.loads(response.read().decode("utf-8") or "{}")
        except urllib.error.HTTPError as error:
            # Every error the server returns has a JSON body worth reading, including 500 run records.
            raw = error.read().decode("utf-8")
            try:
                return error.code, json.loads(raw or "{}")
            except json.JSONDecodeError:
                return error.code, {"error": {"message": raw or error.reason}}

    def _request(
        self,
        route: str,
        payload: Optional[Dict[str, Any]] = None,
        timeout: float = 60.0,
    ) -> Dict[str, Any]:
        status, body = self._raw_request(self.resolve_endpoint(), route, payload, timeout)
        # A failed run is HTTP 500 with a COMPLETE record. Only an envelope without
        # schemaVersion means no run happened.
        if status >= 400 and body.get("schemaVersion") != 1:
            error = body.get("error") or {}
            raise CephableRequestError(
                error.get("message", f"{route} returned HTTP {status}"),
                status,
                error.get("type"),
            )
        return body

    # ── endpoints ────────────────────────────────────────────────────────────────

    def health(self) -> Dict[str, Any]:
        return self._request("/health", timeout=10.0)

    def models(self) -> List[Dict[str, Any]]:
        return self._request("/v1/automate/models", timeout=15.0)["data"]

    def select_model(
        self,
        model_name: Optional[str] = None,
        family: Optional[str] = None,
        size_code: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Pin a model for this app session. Call with no arguments to restore the app's choice."""
        selector = {
            key: value
            for key, value in (("modelName", model_name), ("family", family), ("sizeCode", size_code))
            if value
        }
        return self._request("/v1/automate/models/select", selector, timeout=120.0)

    def cancel(self, force: bool = False) -> Dict[str, Any]:
        """Always safe to call, even when nothing is running."""
        return self._request("/v1/automate/cancel", {"force": force}, timeout=30.0)

    def wait_until_ready(self, timeout: float = 300.0, poll: float = 1.0) -> Dict[str, Any]:
        """The user's own panel runs hold the single inference slot too."""
        deadline = time.monotonic() + timeout
        health: Dict[str, Any] = {}
        while time.monotonic() < deadline:
            health = self.health()
            if not health.get("busy") and health.get("workflowStatus") in READY_STATUSES:
                return health
            time.sleep(poll)
        raise TimeoutError(f"Cephable stayed busy for {timeout}s (last: {health.get('workflowStatus')})")

    def run(
        self,
        prompt: str,
        *,
        task_id: Optional[str] = None,
        timeout_ms: int = 900_000,
        thinking_level: Optional[str] = None,
        additional_workflow_prompt: Optional[str] = None,
        answer_contract: Optional[str] = None,
        include_steps: bool = True,
        include_trace: bool = False,
        include_events: bool = False,
        restrict_to_workspace: bool = False,
        continuation: bool = False,
        selected_skill_ids: Optional[Sequence[str]] = None,
        selected_mcp_server_ids: Optional[Sequence[str]] = None,
        mcp_servers: Optional[List[Dict[str, Any]]] = None,
        client_tools: Optional[List[Dict[str, Any]]] = None,
        hitl_answers: Optional[Dict[str, Union[str, List[str]]]] = None,
        allow_destructive_tools: bool = False,
    ) -> RunResult:
        """Run one Automate task. Blocks until it reaches a terminal state."""
        payload: Dict[str, Any] = {
            "prompt": prompt,
            "timeoutMs": timeout_ms,
            "include": {"steps": include_steps, "trace": include_trace, "events": include_events},
            "restrictToWorkspace": restrict_to_workspace,
            "continuation": continuation,
            "allowDestructiveTools": allow_destructive_tools,
        }
        if task_id:
            payload["taskId"] = task_id
        if thinking_level:
            payload["thinkingLevel"] = thinking_level
        if additional_workflow_prompt:
            payload["additionalWorkflowPrompt"] = additional_workflow_prompt
        if answer_contract:
            payload["answerContract"] = answer_contract
        if selected_skill_ids:
            payload["selectedSkillIds"] = list(selected_skill_ids)
        if selected_mcp_server_ids:
            payload["selectedMcpServerIds"] = list(selected_mcp_server_ids)
        if mcp_servers:
            payload["mcpServers"] = mcp_servers
        if client_tools:
            payload["clientTools"] = client_tools
        if hitl_answers:
            payload["hitlAnswers"] = hitl_answers

        # Keep the client deadline above the server's so its own timeout wins, rather than
        # abandoning a run that is still executing inside Cephable.
        body = self._request("/v1/runs", payload, timeout=timeout_ms / 1000 + 30)
        return RunResult(body)

    def run_or_raise(self, prompt: str, **kwargs: Any) -> RunResult:
        result = self.run(prompt, **kwargs)
        if not result.completed:
            raise CephableRunError(result)
        return result

    def resume_with_tool_results(
        self,
        resume_token: str,
        results: List[Dict[str, Any]],
        timeout: float = 930.0,
    ) -> RunResult:
        """Post results for a parked run and get its next outcome."""
        body = self._request(f"/v1/runs/{resume_token}/tool-results", {"results": results}, timeout=timeout)
        return RunResult(body)

    def run_with_tools(
        self,
        prompt: str,
        handlers: Dict[str, Any],
        *,
        client_tools: Optional[List[Dict[str, Any]]] = None,
        **kwargs: Any,
    ) -> RunResult:
        """
        Run a task with tools this process executes, driving the park/resume loop to completion.

        A parked run holds the app's single inference slot and gets two minutes per round, so a handler
        that raises is reported as a tool error rather than left to time out — the agent adapts or
        explains, which is nearly always better than a dead run.
        """
        definitions = client_tools or [
            {"name": name, "description": f"The {name} tool"} for name in handlers
        ]
        result = self.run(prompt, client_tools=definitions, **kwargs)

        while result.awaiting_tool_results:
            results: List[Dict[str, Any]] = []
            for call in result.tool_calls:
                handler = handlers.get(call["name"])
                if handler is None:
                    results.append({"id": call["id"], "error": f"No handler is registered for {call['name']}"})
                    continue
                try:
                    results.append({"id": call["id"], "result": handler(**(call.get("arguments") or {}))})
                except Exception as error:  # reported to the agent, not raised at the caller
                    results.append({"id": call["id"], "error": str(error)})
            result = self.resume_with_tool_results(result.raw["resumeToken"], results)

        return result
```

---

## Using it

```python
# example.py
from cephable import CephableClient, CephableRequestError, CephableRunError

cephable = CephableClient()

def main() -> int:
    health = cephable.health()
    print(f"Cephable {health['appVersion']} · {health['modelName']} · {health['backend']['accelerator']}")

    cephable.wait_until_ready()

    result = cephable.run_or_raise(
        "Count the PDF files in my Documents folder.",
        task_id="demo-1",
        answer_contract="End with FINAL ANSWER: <integer>",
        timeout_ms=300_000,
    )

    print("answer:", result.value)
    for step in result.steps:
        print(f"  {step['index']}. {step['title']} [{step.get('toolName')}] -> {step['status']}")
    print(f"{result.duration_ms}ms · {(result.usage or {}).get('outputTokens', 0)} output tokens")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except CephableRunError as error:
        print(f"The run did not finish: {error.result.status} {error.result.error_code}")
        for step in error.result.failed_steps:
            print(f"  {step.get('toolName')}: {step.get('resultSummary')}")
        raise SystemExit(1)
    except CephableRequestError as error:
        if error.is_busy:
            print("Cephable is busy with another run.")
        elif error.is_unauthorized:
            print("The access key was rejected. Copy it again from Extensions.")
        else:
            print(f"Request failed ({error.status}): {error}")
        raise SystemExit(1)
```

---

## Structured output with a contract

```python
import json
import re
from typing import TypedDict


class FolderReport(TypedDict):
    fileCount: int
    largestFile: str


def folder_report(cephable: CephableClient, folder: str) -> FolderReport:
    result = cephable.run_or_raise(
        f"Inspect {folder}. Count the files and identify the largest one.",
        answer_contract=(
            'End your reply with one line: '
            'FINAL ANSWER: {"fileCount": <integer>, "largestFile": "<file name>"}'
        ),
    )

    raw = result.value.strip()
    # A small local model will occasionally wrap the value in a code fence.
    raw = re.sub(r"^```(?:json)?\s*", "", raw)
    raw = re.sub(r"\s*```$", "", raw)
    try:
        return json.loads(raw)
    except json.JSONDecodeError as error:
        raise ValueError(f"The assistant did not honor the contract. Answer was: {raw}") from error
```

---

## Giving the agent your own tools

`run_with_tools` drives the whole park/resume loop, so your integration reads like an ordinary call:

```python
def lookup_order(id: str) -> dict:
    order = db.orders.find_by_id(id)
    if order is None:
        raise ValueError(f"No order {id}")   # reported to the agent as a tool error
    return {"status": order.status, "items": order.items, "shippedAt": order.shipped_at}


result = cephable.run_with_tools(
    "Look up order 4471 and draft the customer a reply about the delay.",
    {"lookup_order": lookup_order},
    client_tools=[
        {
            "name": "lookup_order",
            "description": "Fetch an order by its id. Returns status, items and ship date.",
            "parameters": {
                "type": "object",
                "properties": {"id": {"type": "string", "description": "The order id"}},
                "required": ["id"],
            },
        }
    ],
)

print(result.status, result.answer)
```

The agent chains your tools with its own, so `draft_email` can act on what `lookup_order` returned.

To point it at an MCP server you already host instead, the run stays a single call:

```python
result = cephable.run_or_raise(
    "Look up order 4471 and draft the customer a reply about the delay.",
    mcp_servers=[
        {
            "name": "my-app",
            "description": "Order lookup and customer records for the ACME store",
            "transport": "http",
            "url": "http://127.0.0.1:9123/mcp",
            "headers": {"Authorization": f"Bearer {os.environ['INTERNAL_TOKEN']}"},
        }
    ],
)
```

For the LangChain route — `ChatOpenAI(...).bind_tools([...])` against this server, with no Cephable-specific
code at all — see [Custom tools](/docs/automate-http-server/custom-tools).

---

## Workspace round-trip

Stage inputs into the server's workspace folder, run confined to it, then read what came out. This is the pattern for unattended and batch work.

```python
import os
import shutil

def summarize_csvs(cephable: CephableClient, source_dir: str) -> str:
    workspace = cephable.health()["workspace"]
    os.makedirs(workspace, exist_ok=True)

    for name in os.listdir(source_dir):
        if name.endswith(".csv"):
            shutil.copy2(os.path.join(source_dir, name), os.path.join(workspace, name))

    before = set(os.listdir(workspace))

    result = cephable.run_or_raise(
        "Read every .csv in this folder, total the Amount column per file, "
        "and write results.md summarizing them.",
        restrict_to_workspace=True,   # confine file and CLI tools to the workspace
        answer_contract="End with FINAL ANSWER: <grand total as a number>",
        timeout_ms=600_000,
    )

    new_files = sorted(set(os.listdir(workspace)) - before)
    print("new files:", new_files)
    print("paths reported by the run:", result.produced_files)
    return result.value
```

---

## Batch runs, serialized

The app has one inference slot, so a batch has to be sequential. Cancel on interruption so an abandoned run does not block the user.

```python
def run_batch(cephable: CephableClient, prompts: list[str]) -> list[RunResult]:
    results: list[RunResult] = []
    try:
        for index, prompt in enumerate(prompts):
            cephable.wait_until_ready()
            print(f"[{index + 1}/{len(prompts)}] {prompt[:60]}…")
            results.append(
                cephable.run(
                    prompt,
                    task_id=f"batch-{index}",
                    restrict_to_workspace=True,
                    timeout_ms=300_000,
                )
            )
    except KeyboardInterrupt:
        cephable.cancel(force=True)
        raise
    return results
```

---

## `httpx` variant

If you already depend on `httpx`, the transport collapses to a few lines. Note `follow_redirects=False` and a long timeout.

```python
import os
import httpx

client = httpx.Client(
    base_url=os.environ.get("CEPHABLE_ENDPOINT", "http://127.0.0.1:4317"),
    headers={"authorization": f"Bearer {os.environ['CEPHABLE_AUTOMATE_KEY']}"},
    timeout=httpx.Timeout(connect=5.0, read=930.0, write=30.0, pool=5.0),
)

response = client.post(
    "/v1/runs",
    json={
        "prompt": "List the three largest files in my Downloads folder.",
        "include": {"trace": False, "events": False},
    },
)
body = response.json()

if body.get("schemaVersion") == 1:
    print(body["status"], body["answer"])
else:
    print("request failed:", response.status_code, body["error"]["message"])
```

---

## OpenAI SDK variant

For eval harnesses and anything already written against the OpenAI client. Two settings are essential.

```python
import os
from openai import OpenAI

client = OpenAI(
    base_url="http://127.0.0.1:4317/v1",
    api_key=os.environ["CEPHABLE_AUTOMATE_KEY"],
    timeout=930.0,   # agent runs are long; the SDK default is far too short
    max_retries=0,   # a retry either hits 409 or starts a duplicate run
)

completion = client.chat.completions.create(
    model="cephable-agent",
    messages=[{"role": "user", "content": "Summarize the newest file in my Downloads folder."}],
    extra_body={"cephable": {"include": {"trace": False, "events": False}}},
)

print(completion.choices[0].message.content)

record = (completion.model_extra or {}).get("cephable", {})
print(record.get("status"), len(record.get("steps", [])), "steps", record.get("durationMs"), "ms")
```

Only the **last** user message becomes the prompt — earlier turns are not replayed. Fold context into that message, or set `cephable.continuation: True` and send just the new turn. See [OpenAI compatibility](/docs/automate-http-server/openai-compatibility) for the full mapping and the ignored parameters.