CrewAI integration
Five-minute path to wiring STACK into a CrewAI multi-agent crew. Each crew member gets its own STACK agent identity, its own passport, and its own scoped tools — so you can answer the question “which crew member did this?” from a single audit row, and revoking one agent doesn't take down the rest.
1. Install + sign in
pip install crewai getstack
npx -y @getstackrun/cli auth loginSTACK's Python SDK ships on PyPI as getstack. The CLI's auth login stores a refresh token at ~/.stack/credentials.json that the SDK reads automatically, so local development does not need a STACK_API_KEY environment variable. For a first headless deployment, put the operator-wide key in the deployment's secret store only long enough to enroll each agent key, then remove it from the runtime.
2. Register each crew member as a STACK agent
from getstack import Stack
stack = Stack() # reads ~/.stack/credentials.json
researcher = stack.agents.register(
name="researcher",
description="Web research and source-gathering",
accountability_mode="logged", # full audit + detectors, no per-call gate
key_mode="customer_managed",
)
writer = stack.agents.register(
name="writer",
description="Drafts content from researcher findings",
accountability_mode="logged", # full audit + detectors, no per-call gate
key_mode="customer_managed",
)
print(researcher.id, writer.id) # save these for the runtimeEach STACK agent is a separate identity in the audit log, with its own keypair, Passport and scope. Customer-managed key mode is required here because the runtime clients in step 4 authenticate with local agent keys. Compromising one agent's Passport exposes only the authority in that Passport; the other crew members keep separate credentials and grants.
3. Connect the services each agent will need
Connect the upstream services at getstack.run/app/connect. For this example: GitHub for the researcher and Notion for the writer. Each service is connected once at the operator level; then grant each agent exactly the connections it needs. The mission scope in step 4 can only narrow a grant, never substitute for one.
conns = {c["provider"]: c for c in stack.services.list()}
stack.services.grant_agent_access(
agent_id=researcher.id,
service_connection_id=conns["github"]["id"],
scopes=conns["github"]["scopes"],
)
stack.services.grant_agent_access(
agent_id=writer.id,
service_connection_id=conns["notion"]["id"],
scopes=conns["notion"]["scopes"],
)4. Open a mission per agent (in the runtime)
# In your CrewAI runtime — bind a Stack instance per agent so each
# crew member has its own keypair (compromise isolation, scope isolation).
researcher_stack = Stack(agent_id=researcher.id)
writer_stack = Stack(agent_id=writer.id)
# Researcher mission — short-lived, scoped to GitHub only
researcher_mission = researcher_stack.passports.mission(
agent_id=researcher.id,
intent="Gather issue context from acme/sandbox",
services=["github"],
checkpoint_interval="5m",
)
# Writer mission — scoped to Notion only
writer_mission = writer_stack.passports.mission(
agent_id=writer.id,
intent="Draft a Notion page from the researcher's findings",
services=["notion"],
checkpoint_interval="5m",
)stack.passports.mission(...) is a Python context manager — enter it with with and the passport is issued, checkpoints fire automatically, and checkout fires when the block exits. Scope only narrows down a mission's lifetime; nothing the crew does can widen it.
5. STACK-aware tools per agent
Each tool holds the calling agent's passport token and routes the actual upstream call through STACK's proxy.
from crewai.tools import BaseTool
from pydantic import ConfigDict
class StackProxyTool(BaseTool):
"""Generic STACK-proxied tool — supply name, description, service, url."""
model_config = ConfigDict(arbitrary_types_allowed=True)
name: str
description: str
service: str # provider slug, e.g. "notion"
target_url: str # full upstream URL
method: str = "POST"
stack_client: Stack
passport_token: str
def _run(self, **kwargs) -> dict:
response = self.stack_client.proxy.request(
service=self.service,
url=self.target_url,
method=self.method,
body=kwargs or None,
passport_token=self.passport_token,
)
if not response.ok():
raise RuntimeError(f"{self.service} call failed: {response.status}")
return response.body or {}stack.proxy.request(...) is the canonical proxy call. The proxy validates the URL (rejects relative paths), injects the credential server-side, runs scope + constraint checks against the passport, and returns a structured response with .status, .body, and .headers.
Durable non-GET tools should persist the generated operation ID through on_operation_id, reuse it only for an unchanged retry, and use stack.proxy.get_operation() after response loss. See Proxy operation recovery.
6. Wire the crew
from crewai import Agent, Task, Crew
with researcher_mission as r_run, writer_mission as w_run:
researcher_agent = Agent(
role="Researcher",
goal="Summarize the open issues in acme/sandbox",
backstory="Use repository data and report issue URLs.",
tools=[
StackProxyTool(
name="search",
description="List issues in acme/sandbox",
service="github",
target_url="https://api.github.com/repos/acme/sandbox/issues",
method="GET",
stack_client=researcher_stack,
passport_token=r_run.token,
),
],
)
writer_agent = Agent(
role="Writer",
goal="Draft a 500-word brief in Notion citing the researcher's sources",
backstory="Clear prose, faithful to the brief",
tools=[
StackProxyTool(
name="create_notion_page",
description="Create a Notion page",
service="notion",
target_url="https://api.notion.com/v1/pages",
method="POST",
stack_client=writer_stack,
passport_token=w_run.token,
),
],
)
research_task = Task(
description="Summarize open issues in acme/sandbox and include their URLs.",
agent=researcher_agent,
expected_output="A list of 3 source URLs and a one-paragraph summary",
)
draft_task = Task(
description="Draft a Notion page summarizing the research.",
agent=writer_agent,
expected_output="A Notion page id",
)
crew = Crew(agents=[researcher_agent, writer_agent], tasks=[research_task, draft_task])
crew.kickoff()6b. Pre-execution approval gate
Register a crew member with accountability_mode="enforced" and this gate becomes mandatory for it: the proxy refuses any call without an approval id. Enforced-mode agents route Intents through STACK's approval queue before executing. Submit, wait, then thread the returned approval id on the producer call. Rejection auto-revokes the passport.
import time
class GatedStripeRefundTool(BaseTool):
model_config = ConfigDict(arbitrary_types_allowed=True)
name: str = "issue_refund"
description: str = "Refund a Stripe charge after operator approval"
stack_client: Stack
agent_id: str
passport_token: str
def _run(self, charge_id: str, amount_cents: int, reason: str) -> dict:
result = self.stack_client.intents.submit_and_wait(
intent={
"type": "intent_claim",
"intent_type": "http_call",
"agent_id": self.agent_id,
"named_intent": "stripe.create_refund",
"target": "stripe",
"action": "POST /v1/refunds",
"parameters": {
"url": "https://api.stripe.com/v1/refunds",
"method": "POST",
"body": {"charge": charge_id, "amount": amount_cents},
},
"estimated_cost": {"wallet_cents": 0, "tokens": None, "gas_gwei": None},
"accountability": "enforced",
"reason": reason,
"requires": [],
"user_subject": None,
"mission_ref": None,
"submitted_at": int(time.time() * 1000),
},
passport_token=self.passport_token,
timeout_seconds=300,
)
if result["final"]["status"] != "approved":
raise RuntimeError(f"Refund blocked: {result['final']['status']}")
response = self.stack_client.proxy.request(
service="stripe",
url="https://api.stripe.com/v1/refunds",
method="POST",
body={"charge": charge_id, "amount": amount_cents},
passport_token=self.passport_token,
approval_id=result["final"]["id"],
)
return response.body or {}The gate checks the call shape against the approved Intent (service, method, URL host, body). Mismatch returns 403 with metadata.gate_reason="call_mismatch". stack.intents.simulate(...) is the pre-check that runs without a human round-trip.
7. (Optional) Structured handoff via drop-offs
Free-form text passing between crew agents is fragile. Drop-offs let the producer declare a JSON Schema; the consumer collects once; the payload is destroyed after collection or expiry.
# Researcher creates a drop-off addressed to the writer
dropoff = researcher_stack.dropoffs.create(
from_agent_id=researcher.id,
to_agent_id=writer.id,
schema={
"type": "object",
"required": ["sources", "summary"],
"properties": {
"sources": {"type": "array", "items": {"type": "string"}},
"summary": {"type": "string", "maxLength": 4000},
},
},
ttl_seconds=600,
)
# Researcher deposits — agent_id is the depositing agent
researcher_stack.dropoffs.deposit(
dropoff_id=dropoff["id"],
data={"sources": ["https://...", "https://..."], "summary": "..."},
agent_id=researcher.id,
)
# Writer collects — agent_id is the collecting agent
findings = writer_stack.dropoffs.collect(
dropoff_id=dropoff["id"],
agent_id=writer.id,
)Both deposit and collect require agent_id — STACK uses it to enforce the from/to addressing on the drop-off. Schema validation runs at deposit; the consumer is guaranteed schema-conforming data or no data at all.
8. Revocation
# Revoke one agent's Passport
stack.passports.revoke(researcher_mission.passport.jti, reason="off-script")
# Revoke every active passport for an agent
# (no SDK helper — call the API directly)
stack._client.post(f"/v1/passports/revoke-agent/{researcher.id}")See drop-offs concept and the passport API reference for the full contracts on cross-agent flows and revocation.
Per-agent controls
- Each crew member is a separate STACK identity — audit rows tell you which agent did what
- Per-agent passports mean per-agent scope, per-agent revocation, per-agent detector signals
- Drop-offs replace free-form text passing with schema-validated, encrypted, one-read handoffs
- Per-agent revocation rejects that agent on its next STACK-verified call; the rest of the crew keeps running