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

bash
pip install crewai getstack
npx -y @getstackrun/cli auth login

STACK'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 -- no STACK_API_KEY env var needed for local dev. (For CI: set STACK_API_KEY the legacy way.)

2. Register each crew member as a STACK agent

python
from getstack import Stack

stack = Stack()  # reads ~/.stack/credentials.json

researcher = stack.agents.register(
    name="researcher",
    description="Web research and source-gathering",
    accountability_mode="enforced",
)
writer = stack.agents.register(
    name="writer",
    description="Drafts content from researcher findings",
    accountability_mode="enforced",
)
print(researcher.id, writer.id)  # save these for the runtime

Each STACK agent is a separate identity in the audit log, with its own passport and its own scope. A leaked credential or a misbehaving crew member only burns the radius you grant that agent.

3. Connect the services each agent will need

Connect the upstream services at getstack.run/services. For this example: a search provider for the researcher and Notion for the writer. Each service is connected once at the operator level; both agents inherit access according to the passport scope you issue them.

4. Open a mission per agent (in the runtime)

python
# 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 search only
researcher_mission = researcher_stack.passports.mission(
    agent_id=researcher.id,
    intent="Gather sources on agent identity standards",
    services=["serpapi"],
    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.

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

6. Wire the crew

python
from crewai import Agent, Task, Crew

with researcher_mission as r_run, writer_mission as w_run:
    researcher_agent = Agent(
        role="Researcher",
        goal="Find 3 high-quality sources on agent identity standards",
        backstory="Concise, factual, sources every claim",
        tools=[
            StackProxyTool(
                name="search",
                description="Search the web via SerpAPI",
                service="serpapi",
                target_url="https://serpapi.com/search.json",
                method="GET",
                stack_client=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=stack,
                passport_token=w_run.token,
            ),
        ],
    )

    research_task = Task(
        description="Find 3 sources on agent identity standards.",
        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

Enforced-mode agents can 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.

python
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:refunds:create",
                "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.

python
# Researcher creates a drop-off addressed to the writer
dropoff = 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
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 = 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. Kill switch

python
# Revoke one agent's passport (sub-60s propagation)
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.

Why bother

  • 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
  • Sub-60-second kill switch per agent — the rest of the crew keeps running
stack | docs