STACK
MENU
DOCS / SDKS AND CLI / PYTHON

Python SDK

The package provides synchronous and asynchronous clients. It requires Python 3.9+ and uses httpx for HTTP requests.

Installation

bash
pip install getstack

Quick Start

Sign in once on your development machine:

bash
npx @getstackrun/cli auth login
python
from getstack import Stack

stack = Stack()
agents = stack.agents.list()

# Open a Mission for one agent and issue its Passport in one call.
started = stack.missions.start(
    agent_id="agt_support",
    mission={"intent_summary": "Post the daily support update", "intent_services": ["slack"]},
)
token = started["passport"]["token"]

# Completing the Mission revokes its Passports.
stack.missions.complete(started["mission_id"])

Stack() creates an API client and can use the local OAuth profile written by the CLI. A fleet worker uses Stack(agent_id=agent_id) to sign requests with its locally stored key. AsyncStack exposes the same methods with await.

Fleet setup

Connect the service first and name an enrolled provisioner in the policy. wait_for_approval opens the returned approval URL and waits for the operator's decision. create_worker returns Stack(agent_id=agent_id), or AsyncStack for the async example. Enrollment stores the worker's key locally and needs no separate operator approval.

python
from getstack.errors import StackError


def setup_fleet(*, operator, provisioner, policy, policy_key, worker_name,
                enrollment_key, create_worker, wait_for_approval):
    try:
        activated = operator.fleets.activate(policy, idempotency_key=policy_key)
    except StackError as error:
        approval = error.details
        if (error.code != "GOVERNANCE_APPROVAL_REQUIRED"
                or not approval.get("approval_id") or not approval.get("approval_url")):
            raise
        wait_for_approval(approval)
        activated = operator.fleets.activate(policy, idempotency_key=policy_key,
            governance_approval_id=approval["approval_id"])
    enrolled = provisioner.fleets.enroll_local(activated["id"],
        name=worker_name, idempotency_key=enrollment_key)
    worker = create_worker(enrolled["agent_id"])
    session = worker.fleets.session()
    return {"policy_id": activated["id"], "agent_id": enrolled["agent_id"],
            "worker": worker, "session": session}
python
from getstack.errors import StackError


async def setup_fleet(*, operator, provisioner, policy, policy_key, worker_name,
                      enrollment_key, create_worker, wait_for_approval):
    try:
        activated = await operator.fleets.activate(policy, idempotency_key=policy_key)
    except StackError as error:
        approval = error.details
        if (error.code != "GOVERNANCE_APPROVAL_REQUIRED"
                or not approval.get("approval_id") or not approval.get("approval_url")):
            raise
        await wait_for_approval(approval)
        activated = await operator.fleets.activate(policy, idempotency_key=policy_key,
            governance_approval_id=approval["approval_id"])
    enrolled = await provisioner.fleets.enroll_local(activated["id"],
        name=worker_name, idempotency_key=enrollment_key)
    worker = create_worker(enrolled["agent_id"])
    session = await worker.fleets.session()
    return {"policy_id": activated["id"], "agent_id": enrolled["agent_id"],
            "worker": worker, "session": session}

Fleet Mission workflow

python
import hashlib
import json
from getstack.errors import StackError


def run_fleet_mission_example(stack, request, provider_url="https://slack.com/api/chat.postMessage"):
    started = stack.missions.start(request)
    repeated = stack.missions.start(request)
    assert repeated["mission_id"] == started["mission_id"]
    operation_id = hashlib.sha256(json.dumps([started["mission_id"], "support-update"]).encode()).hexdigest()
    response = stack.proxy.request("slack", provider_url, "POST",
        body={"channel": "C123", "text": "Support update"},
        passport_token=started["passport"]["token"], authority_request_id=operation_id)
    assert response.status == 200
    try:
        stack.proxy.request("slack", provider_url, "POST", body={"channel": "OTHER", "text": "Support update"},
            passport_token=started["passport"]["token"], authority_request_id=operation_id + "_denied")
    except StackError as error:
        assert error.status_code == 403
    else:
        raise AssertionError("Expected channel restriction")
    before = stack.missions.get(started["mission_id"])
    renewed = stack.missions.renew(started["mission_id"])
    after = stack.missions.get(renewed["mission_id"])
    assert renewed["mission_id"] == started["mission_id"]
    assert after["fleet"]["limits"] == before["fleet"]["limits"]
    assert after["fleet"]["usage"] == before["fleet"]["usage"]
    completed = stack.missions.complete(started["mission_id"], output={"done": True})
    assert completed["terminal_state"] == "completed_successful"
    stopped = stack.missions.start({**request, "idempotency_key": request["idempotency_key"] + "-stop"})
    stack.missions.revoke(stopped["mission_id"], reason="operator_initiated_clean")
    try:
        stack.missions.renew(stopped["mission_id"])
    except StackError:
        pass
    else:
        raise AssertionError("Expected revoked Mission")
    return {"mission_id": started["mission_id"], "operation_id": operation_id, "completed": True, "revoked": True}
python
import hashlib
import json
from getstack.errors import StackError


async def run_fleet_mission_example(stack, request, provider_url="https://slack.com/api/chat.postMessage"):
    started = await stack.missions.start(request)
    repeated = await stack.missions.start(request)
    assert repeated["mission_id"] == started["mission_id"]
    operation_id = hashlib.sha256(json.dumps([started["mission_id"], "support-update"]).encode()).hexdigest()
    response = await stack.proxy.request("slack", provider_url, "POST",
        body={"channel": "C123", "text": "Support update"},
        passport_token=started["passport"]["token"], authority_request_id=operation_id)
    assert response.status == 200
    try:
        await stack.proxy.request("slack", provider_url, "POST", body={"channel": "OTHER", "text": "Support update"},
            passport_token=started["passport"]["token"], authority_request_id=operation_id + "_denied")
    except StackError as error:
        assert error.status_code == 403
    else:
        raise AssertionError("Expected channel restriction")
    before = await stack.missions.get(started["mission_id"])
    renewed = await stack.missions.renew(started["mission_id"])
    after = await stack.missions.get(renewed["mission_id"])
    assert renewed["mission_id"] == started["mission_id"]
    assert after["fleet"]["limits"] == before["fleet"]["limits"]
    assert after["fleet"]["usage"] == before["fleet"]["usage"]
    completed = await stack.missions.complete(started["mission_id"], output={"done": True})
    assert completed["terminal_state"] == "completed_successful"
    stopped = await stack.missions.start({**request, "idempotency_key": request["idempotency_key"] + "-stop"})
    await stack.missions.revoke(stopped["mission_id"], reason="operator_initiated_clean")
    try:
        await stack.missions.renew(stopped["mission_id"])
    except StackError:
        pass
    else:
        raise AssertionError("Expected revoked Mission")
    return {"mission_id": started["mission_id"], "operation_id": operation_id, "completed": True, "revoked": True}

stack.missions starts and manages server-side Missions. The with stack.passports.mission(agent_id, intent, services) context manager opens a Mission and its Passport in one call (POST /v1/missions/start), completes the Mission on exit and revokes it if the block raises. Pass mission_id="ms_..." to bind an existing Mission instead; the helper then leaves that Mission open.

Services

stack.fleets and stack.missions

python
stack.fleets.list()
stack.fleets.list_page(page=1, limit=20)
stack.fleets.get(policy_id)
stack.fleets.membership(agent_id)
stack.fleets.members_page(policy_id, page=1, limit=20)
stack.fleets.activate(policy, idempotency_key=key, governance_approval_id=approval_id)
stack.fleets.revise(policy_id, policy, expected_revision=revision, idempotency_key=key, governance_approval_id=approval_id)
stack.fleets.members(policy_id)
stack.fleets.revoke_member(policy_id, agent_id)
stack.fleets.revoke(policy_id)
stack.fleets.enroll_local(policy_id, name=name, idempotency_key=key)
stack.fleets.enroll_workload_local(policy_id, name=name, idempotency_key=key)
stack.fleets.session(fleet_policy_id)
stack.missions.start(agent_id="agt_...", mission={"intent_summary": "...", "intent_services": [...]}, passport={...})
stack.missions.start(fleet_request)  # fleet form: dict with fleet_policy_id
stack.missions.open("summary", agent_id="agt_...", intent_services=[...], expires_in_seconds=3600)
stack.missions.renew(mission_id)
stack.missions.complete(mission_id, output={"done": True})
stack.missions.revoke(mission_id, reason="operator_initiated_clean")

The operator form of missions.start returns a dict with mission_id, claim_id, expires_at and passport (token, jti, expires_at). passport is optional and takes issue fields such as scopes, ttl_seconds and intent. missions.open assigns the Mission to agent_id when given. Completing or revoking a Mission revokes every Passport bound to it, for any revoke reason.

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.passports

python
mission = stack.missions.open("Post the daily support update",
                              agent_id="agt_...", intent_services=["slack"])
passport = stack.passports.issue(
    agent_id="agt_...",
    mission_ids=[mission["mission_id"]],
    authority_binding_ids=["abn_..."],
)
print(passport.token)

mission_ids is required and holds exactly one live Mission ID; the SDK raises ValueError otherwise. The Mission must be assigned to the agent or unassigned. The Passport never outlives its Mission, and refresh and delegation fail once the Mission is completed, revoked or expired.

The issue helper attaches active bindings and the Passport receives opaque authority references. Authority management and the native receiver protocol use the Authority REST API. The async client accepts the same authority_binding_ids argument.

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()
# Start a provider shown as available for this deployment.
provider_key = providers[0].key
session = stack.identity.initiate_verification(provider_key)
stack.identity.complete_verification(provider_key, session.session_ref)
claims = stack.identity.list_claims()
stack.identity.revoke_claim("clm_...")
stack.identity.replace_service_requirement(service_id, requirement_or_none)

Authentication Methods

python
import os

# Local OAuth profile created by the CLI
stack = Stack()

# Operator API key from the runtime secret store
stack = Stack(api_key=os.environ["STACK_API_KEY"])

# Existing dashboard session
stack = Stack.from_session(session_token)

stack.dropoffs / stack.proxy / stack.scan / stack.delivery_methods / stack.security_events

python
# Drop-offs
dof = stack.dropoffs.create(from_agent_id="agt_a", to_agent_id="agt_b", schema={...})
# Each agent passes the Passport from its own stack.missions.start(...).
stack.dropoffs.deposit(dof["id"], data={...}, agent_id="agt_a", passport_token=producer_token)
stack.dropoffs.collect(dof["id"], agent_id="agt_b", passport_token=consumer_token)

# Proxy
resp = stack.proxy.request(service="slack", method="POST",
                           url="https://slack.com/api/chat.postMessage",
                           body={"channel": "C0123", "text": "hi"},
                           passport_token=passport.token,
                           on_operation_id=save_operation_id)
operation = stack.proxy.get_operation(resp.operation_id,
                                      passport_token=passport.token,
                                      include_response=True)

# 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

# Delivery methods (email, SMS or webhook; each is verified before it delivers)
dest = stack.delivery_methods.add("webhook", "https://…")
stack.delivery_methods.send_verification_code(dest["id"])
stack.delivery_methods.verify(dest["id"], code="123456")
stack.delivery_methods.test(dest["id"])

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

A notification rule routes events to verified delivery methods. The Python client has no helper for creating one; send POST /v1/notifications/channels with destination_ids, events and an optional min_severity. See Notifications API.

Non-GET proxy calls generate an operation ID before network I/O. Durable workers should use on_operation_id to persist it, then reuse authority_request_id only for the unchanged request. Sync and async clients both provide get_operation() for recovery. See Proxy operation recovery.

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

# 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. The Python client has no export helper; use GET /v1/audit/export for date-range exports.

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")
    mission = await stack.missions.open("...", agent_id=agent.id, intent_services=[...])
    passport = await stack.passports.issue(agent_id=agent.id, mission_ids=[mission["mission_id"]],
                                           intent="...", services=[...])
    # ...
    await stack.missions.complete(mission["mission_id"])  # also revokes the Passport

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")
← PREV: JAVASCRIPT AND TYPESCRIPTNEXT: CLI COMMAND REFERENCE
STACKONE RUNTIME BOUNDARY FOR AI AGENTS.
BUILT IN STOCKHOLM.
$npx @getstackrun/cli setup
ATTACK OUR AGENT
PRODUCTPRODUCTPRICINGFOR PLATFORMSINTEGRATIONS
DEVELOPERSDOCSACCEPT AGENTSSIGN IN
TRUSTSECURITYAGENT SECURITYATTACK USEU AI ACT ART. 14
COMPANYCONTACTPRIVACYTERMS
© 2026 STACK · Stockholm, SwedenHELLO@GETSTACK.RUN
stack | Docs