Skip to content

Python Integration

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.

from agents.hub_tool import plan
result = plan(
command="Get-Process python",
note="Checking running Python processes",
cwd=None,
timeout_s=300
)
ParameterTypeDefaultDescription
commandstr(required)PowerShell command to queue
notestr|NoneNoneHuman-readable context for the approver
cwdstr|NoneNoneWorking directory (defaults to server’s cwd)
timeout_sint300Max execution time in seconds
{
"pending_id": "20250115_143022_a1b2c3d4",
"queued": True,
"command": "Get-Process python"
}

The SDK reads these from the environment:

VariableDefaultDescription
AGENT_HUB_BASEhttp://127.0.0.1:8787API base URL
AGENT_HUB_TOKEN(empty)Authentication token

The full SDK is intentionally minimal:

import os
import 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()
import os
os.environ["AGENT_HUB_TOKEN"] = "your-token"
from agents.hub_tool import plan
# Agent decides to check system status
result = 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 API
import httpx
from 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?")
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']}")

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()