Python

pip install httpx sseclient-py

import httpx

BASE_URL = "https://{project}.polpo.cloud/v1"
API_KEY = "sk_live_..."

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}

client = httpx.Client(base_url=BASE_URL, headers=headers)

import httpx
import json

def chat_stream(agent: str, messages: list[dict]):
    with httpx.stream(
        "POST",
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json={
            "agent": agent,
            "messages": messages,
            "stream": True,
        },
    ) as response:
        for line in response.iter_lines():
            if line.startswith("data: ") and line != "data: [DONE]":
                chunk = json.loads(line[6:])
                delta = chunk["choices"][0].get("delta", {}).get("content", "")
                if delta:
                    print(delta, end="", flush=True)
        print()

chat_stream("backend-dev", [
    {"role": "user", "content": "Explain the auth module"}
])

response = client.post("/chat/completions", json={
    "agent": "backend-dev",
    "messages": [{"role": "user", "content": "Explain the auth module"}],
    "stream": False,
})

data = response.json()
print(data["choices"][0]["message"]["content"])

response = client.get("/agents")
agents = response.json()["data"]

for agent in agents:
    print(f"{agent['name']} — {agent.get('role', '')}")

response = client.post("/tasks", json={
    "title": "Refactor the payment module",
    "description": "Refactor the module without changing its public API.",
    "assignTo": "backend-dev",
    "group": "sprint-3",
})

task = response.json()["data"]
print(f"Created: {task['id']}")

response = client.get("/missions")
missions = response.json()["data"]

for m in missions:
    print(f"{m['name']} — {m['status']}")

response = client.post(f"/missions/{mission_id}/execute")
result = response.json()["data"]
print(f"Started {len(result['tasks'])} tasks")

import sseclient
import requests

url = f"{BASE_URL}/events"
response = requests.get(url, headers=headers, stream=True)
events = sseclient.SSEClient(response)

for event in events:
    if event.data:
        data = json.loads(event.data)
        print(f"[{event.event}] {data}")

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/webhooks/polpo", methods=["POST"])
def handle_webhook():
    event = request.json

    if event["event"] == "task:transition":
        print(f"Task {event['data']['taskId']} → {event['data']['to']}")
    elif event["event"] == "completion:finished":
        print(f"Completion finished: {event['data'].get('taskId')}")

    return jsonify({"ok": True})

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/webhooks/polpo")
async def handle_webhook(request: Request):
    event = await request.json()

    match event["event"]:
        case "task:transition":
            print(f"Task {event['data']['taskId']} → {event['data']['to']}")
        case "completion:finished":
            print("Completion finished")

    return {"ok": True}

# Read shared memory
response = client.get("/memory")
content = response.json()["data"]["content"]

# Save shared memory
client.put("/memory", json={"content": "Updated context..."})

# Per-agent memory
response = client.get("/memory/agent/backend-dev")
client.put("/memory/agent/backend-dev", json={"content": "Prefers functional patterns"})