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.
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
# 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
# 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
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:
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:
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.
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.
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.
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.
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.
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 for the full mapping and the ignored parameters.