The package provides synchronous and asynchronous clients. It requires Python 3.9+ and uses httpx for HTTP requests.
pip install getstackSign in once on your development machine:
npx @getstackrun/cli auth loginfrom getstack import Stack
stack = Stack()
agents = stack.agents.list()Stack() creates an API client. During local development, it can read the OAuth profile written by the CLI. In CI or production, pass the operator API key from the runtime's secret store and treat it as a full-account credential.
stack.agents.register("name", description="optional")
stack.agents.list()
stack.agents.get("agt_...")
stack.agents.update("agt_...", status="suspended")
stack.agents.delete("agt_...")passport = stack.passports.issue(
agent_id="agt_...",
authority_binding_ids=["abn_..."],
)
print(passport.token)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.
# 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_...")providers = stack.identity.list_providers()
# Start a provider shown as available for this deployment.
provider_key = providers[0].provider_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)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)# Drop-offs
dof = stack.dropoffs.create(from_agent_id="agt_a", to_agent_id="agt_b", schema={...})
stack.dropoffs.deposit(dof["id"], data={...}, agent_id="agt_a")
stack.dropoffs.collect(dof["id"], agent_id="agt_b")
# 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)# 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.
For async applications, import AsyncStack. All service methods are awaitable; the service surface mirrors the sync client exactly.
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)from getstack import Stack, NotFoundError, ForbiddenError
try:
agent = stack.agents.get("agt_nonexistent")
except NotFoundError:
print("Agent not found")
except ForbiddenError:
print("Access denied")