Python SDK

Sync and async clients. Requires Python 3.9+. Uses httpx under the hood.

Installation

bash
pip install getstack

Quick Start

python
from getstack import Stack

stack = Stack(api_key="sk_live_...")

# Register an agent
agent = stack.agents.register("my-agent")

# Issue a passport with mission context
with stack.passports.mission(
    agent_id=agent.id,
    intent="Process invoices",
    services=["slack", "stripe"],
) as mission:
    mission.log("slack", "read_channel", "#invoices")
    cred = stack.credentials.get("slack")
    mission.log("stripe", "create_invoice")

Services

stack.agents

python
stack.agents.register("name", description="optional")
stack.agents.list()
stack.agents.get("agt_...")
stack.agents.update("agt_...", status="suspended")
stack.agents.delete("agt_...")

stack.skills

python
# Publishing
stack.skills.publish(
    name="my-skill",
    description="...",
    input_schema={...},
    output_schema={...},
    execution_mode="sealed",
)

# Browsing
skills = stack.skills.browse(query="financial analysis", tags="finance")

# Invoking
result = stack.skills.invoke(
    "skl_...",
    agent_id="agt_...",
    input={"data": "..."},
)

# Polling
completed = stack.skills.poll(result.id, timeout_seconds=60)

# Requests
stack.skills.post_request(description="I need...", tags=["finance"])
stack.skills.suggest_composition("sreq_...")

stack.identity

python
providers = stack.identity.list_providers()
session = stack.identity.initiate_verification("persona")
stack.identity.complete_verification("persona", session.session_ref)
claims = stack.identity.list_claims()
stack.identity.revoke_claim("clm_...")

Authentication Methods

python
# API key (default)
stack = Stack(api_key="sk_live_...")

# Session JWT
stack = Stack.from_session("eyJ...")

# OAuth
stack = Stack.from_oauth(
    client_id="...",
    client_secret="...",
    access_token="...",
    refresh_token="...",
)

stack.dropoffs / stack.proxy / stack.notifications / stack.security_events

python
# Drop-offs
dof = stack.dropoffs.create(from_agent="agt_a", to_agent="agt_b", schema={...})
stack.dropoffs.deposit(dof.id, payload={...})
stack.dropoffs.collect(dof.id)

# Proxy
resp = stack.proxy.request(service="slack", method="POST",
                           url="https://slack.com/api/chat.postMessage",
                           body={"channel": "C0123", "text": "hi"})

# Content scan — run prompt-injection detector on retrieved content
result = stack.scan.scan(
    content=email_body,
    context="email",
    source=sender_address,
)
if result["verdict"] == "critical":
    raise RuntimeError(f"Indirect injection caught: {result['match']['pattern_id']}")
stack.scan.usage()  # monthly quota

# Notifications
stack.notifications.add_destination(channel_type="webhook", destination="https://…")
stack.notifications.verify_destination(dest_id, code="123456")
stack.notifications.create_rule(destination_ids=[dest_id], events=["passport.flagged"],
                                 min_severity="warning")

# Security events
events = stack.security_events.list(limit=50)
stack.security_events.resolve(event_id)

stack.audit / stack.reviews

python
# Tail recent entries — newest-first. Pass since (epoch ms) for
# incremental polling; pass agent_id or passport_jti to narrow.
data = stack.audit.list(limit=20, agent_id="agt_support")
for entry in data["entries"]:
    print(entry["action"], entry["outcome"])
watermark = data["max_timestamp"]

# Anchor the chain head externally to prove no row was rewritten later.
head = stack.audit.chain_head()

# Walk the chain and verify every entry's hash + link integrity.
result = stack.audit.verify_chain()
assert result["valid"], result.get("first_break")

# Full date-range export (NDJSON / CSV / JSON)
rows = stack.audit.export(format="ndjson", limit=10000)

# Checkout reviews
pending = stack.reviews.list(status="flagged")
stack.reviews.decide(checkout_id="cout_abc", decision="approved",
                    notes="notion use was ancillary")

Cascade revokes write one passport.revoke entry plus onepassport.revoke_cascade per child — filter bypassport_jti= to see the chain for a single passport.

Async client

For async applications, import AsyncStack. All service methods are awaitable; the service surface mirrors the sync client exactly.

python
from getstack import AsyncStack

stack = AsyncStack(api_key="sk_live_...")

async with stack:
    agent = await stack.agents.register("my-agent")
    passport = await stack.passports.issue(agent_id=agent.id, intent="...", services=[...])
    # ...
    await stack.passports.revoke(passport.jti)

Error Handling

python
from getstack import Stack, NotFoundError, ForbiddenError

try:
    agent = stack.agents.get("agt_nonexistent")
except NotFoundError:
    print("Agent not found")
except ForbiddenError:
    print("Access denied")
stack | docs