Python Integration
Overview
Section titled “Overview”The Provara Python SDK (agents/hub_tool.py) provides a single function to queue commands for human approval. It’s designed to be the bridge between any Python-based AI agent and the Provara approval pipeline.
The plan() Function
Section titled “The plan() Function”from agents.hub_tool import plan
result = plan( command="Get-Process python", note="Checking running Python processes", cwd=None, timeout_s=300)Parameters
Section titled “Parameters”| Parameter | Type | Default | Description |
|---|---|---|---|
command | str | (required) | PowerShell command to queue |
note | str|None | None | Human-readable context for the approver |
cwd | str|None | None | Working directory (defaults to server’s cwd) |
timeout_s | int | 300 | Max execution time in seconds |
Return Value
Section titled “Return Value”{ "pending_id": "20250115_143022_a1b2c3d4", "queued": True, "command": "Get-Process python"}Environment Variables
Section titled “Environment Variables”The SDK reads these from the environment:
| Variable | Default | Description |
|---|---|---|
AGENT_HUB_BASE | http://127.0.0.1:8787 | API base URL |
AGENT_HUB_TOKEN | (empty) | Authentication token |
Implementation
Section titled “Implementation”The full SDK is intentionally minimal:
import osimport httpx
BASE = os.environ.get("AGENT_HUB_BASE", "http://127.0.0.1:8787")TOKEN = os.environ.get("AGENT_HUB_TOKEN", "")
def plan(command, note=None, cwd=None, timeout_s=300): print(f"[QUEUE] {command}") r = httpx.post( f"{BASE}/plan", headers={"X-Agent-Token": TOKEN}, json={ "command": command, "note": note, "cwd": cwd, "timeout_s": timeout_s }, timeout=15, ) r.raise_for_status() return r.json()Usage Patterns
Section titled “Usage Patterns”Basic Agent Loop
Section titled “Basic Agent Loop”import osos.environ["AGENT_HUB_TOKEN"] = "your-token"
from agents.hub_tool import plan
# Agent decides to check system statusresult = plan( command="Get-Service | Where-Object Status -eq Running | Measure-Object", note="Counting running services")print(f"Command queued: {result['pending_id']}")# Human reviews and approves/denies via UI or APIWith Error Handling
Section titled “With Error Handling”import httpxfrom agents.hub_tool import plan
try: result = plan( command="Get-Process python", note="diagnostic check" ) print(f"Queued: {result['pending_id']}")except httpx.HTTPStatusError as e: if e.response.status_code == 401: print("Authentication failed — check AGENT_HUB_TOKEN") elif e.response.status_code == 429: print("Rate limited — wait and retry") else: print(f"API error: {e.response.status_code}")except httpx.ConnectError: print("Cannot reach Provara server — is it running?")Batch Command Queueing
Section titled “Batch Command Queueing”commands = [ ("whoami", "identity check"), ("hostname", "hostname check"), ("Get-Date", "timestamp check"),]
pending_ids = []for cmd, note in commands: result = plan(command=cmd, note=note) pending_ids.append(result["pending_id"]) print(f"Queued {cmd}: {result['pending_id']}")Building a Custom Client
Section titled “Building a Custom Client”If you need more control, build a custom client using httpx:
import httpx
class ProvaraClient: def __init__(self, base_url="http://127.0.0.1:8787", token=""): self.base = base_url.rstrip("/") self.headers = {"X-Agent-Token": token} self.client = httpx.Client(timeout=15)
def plan(self, command, **kwargs): r = self.client.post( f"{self.base}/plan", headers=self.headers, json={"command": command, **kwargs} ) r.raise_for_status() return r.json()
def pending(self, limit=50): r = self.client.get( f"{self.base}/pending", headers=self.headers, params={"limit": limit} ) r.raise_for_status() return r.json()
def status(self): r = self.client.get(f"{self.base}/status") r.raise_for_status() return r.json()Next Steps
Section titled “Next Steps”- AutoGen Integration — Use Provara with Microsoft AutoGen
- Custom Agents — Design patterns for agent development