Skip to main content

Quickstart: Zero-Trust Security via Access Policies

Enable authorization for CrewAI with access policies.

This quickstart demonstrates how to secure agent-to-agent communication using Catalyst access policies. You'll run two apps — a caller that invokes a server running a CrewAI pipeline — and then apply a deny policy to block unauthorized access.

note

This tutorial uses Catalyst Cloud to get started quickly with zero infrastructure setup. Catalyst Cloud uses a tunnel to route service invocation requests from the cloud to your local dev machine. For production or on-premises requirements, Diagrid also offers a self-hosted enterprise option where traffic stays within your private network.

You will learn how to:

  • Use Catalyst service invocation for agent-to-agent communication
  • Apply an access control configuration to enforce zero-trust authorization
  • Observe a 403 PermissionDenied response when a caller is blocked by policy

Your apps run entirely on your local machine — you are not deploying code to Catalyst. diagrid dev run opens a secure tunnel between Catalyst Cloud and your local processes and registers each app as a Dapr App ID. When the caller sends a service invocation request, it travels up to Catalyst Cloud (not directly to the server). Catalyst evaluates the access policy there, in the cloud: if the policy allows it, the request is forwarded back down through the tunnel to the local server; if the policy denies it, Catalyst returns a 403 PermissionDenied immediately and the request never reaches your local server process.

1. Prerequisites​

2. Log in to Catalyst​

diagrid login

Confirm your identity:

diagrid whoami

3. Create Project Files​

Create a new directory for the quickstart:

mkdir crewai-access-policies && cd crewai-access-policies
mkdir caller server

server/app.py​

Create server/app.py with the following content. This defines a simple CrewAI crew exposed via a /research HTTP endpoint:

"""Server app: exposes a CrewAI pipeline via HTTP."""
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel


# ── FastAPI ──────────────────────────────────────────────────
app = FastAPI()


class RunRequest(BaseModel):
    topic: str


@app.post("/research")
async def research(req: RunRequest):
    from crewai import Agent, Task, Crew

    researcher = Agent(
        role="Research Analyst",
        goal="Find and summarize research on a given topic",
        backstory="You are an experienced research analyst.",
        allow_delegation=False,
        verbose=True,
    )
    task = Task(
        description=f"Research the topic: {req.topic}. Provide a brief summary.",
        expected_output="A short research summary.",
        agent=researcher,
    )
    crew = Crew(agents=[researcher], tasks=[task], verbose=True)
    result = crew.kickoff()
    return {"result": str(result)}


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8002)

caller/app.py​

Create caller/app.py with the following content. This app invokes the server's /research endpoint through Catalyst service invocation:

"""Caller app: invokes the server app through Catalyst service invocation."""
import os

import httpx
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# Catalyst injects these environment variables via `diagrid dev run`
DAPR_HTTP_ENDPOINT = os.getenv("DAPR_HTTP_ENDPOINT", "http://localhost:3500")
DAPR_API_TOKEN = os.getenv("DAPR_API_TOKEN", "")


class CallRequest(BaseModel):
    topic: str


@app.post("/call-server")
async def call_server(req: CallRequest):
    """Call the server app's /research endpoint via Catalyst service invocation."""
    url = f"{DAPR_HTTP_ENDPOINT}/v1.0/invoke/server/method/research"
    headers = {
        "dapr-api-token": DAPR_API_TOKEN,
        "Content-Type": "application/json",
    }
    print(f">>> Calling server via Catalyst: {url}", flush=True)
    async with httpx.AsyncClient() as client:
        resp = await client.post(url, json={"topic": req.topic}, headers=headers, timeout=30)
    print(f">>> Response status: {resp.status_code}", flush=True)
    print(f">>> Response body: {resp.text}", flush=True)
    return {"status_code": resp.status_code, "body": resp.json() if resp.status_code == 200 else resp.text}


@app.get("/health")
async def health():
    return {"status": "ok"}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

The caller uses DAPR_HTTP_ENDPOINT and DAPR_API_TOKEN, which are automatically injected by diagrid dev run. The service invocation URL follows the pattern:

{DAPR_HTTP_ENDPOINT}/v1.0/invoke/{target-app-id}/method/{endpoint}

dev.yaml​

Create dev.yaml in the project root:

version: 1
common:
  appLogDestination: console
apps:
  - appID: caller
    appDirPath: ./caller
    command: [".venv/bin/python", "app.py"]
  - appID: server
    appDirPath: ./server
    appPort: 8002
    command: [".venv/bin/python", "app.py"]
tip

The appPort field on the server tells Catalyst which local port the app listens on. This is required for apps that receive service invocation requests. The caller does not need appPort because it only sends requests.

4. Configure API Key​

export OPENAI_API_KEY="your-openai-api-key"

5. Install Dependencies​

python3 -m venv .venv
source .venv/bin/activate
pip install crewai fastapi uvicorn httpx

Create symlinks so both apps share the same virtual environment:

ln -s ../.venv caller/.venv
ln -s ../.venv server/.venv

6. Run with Default Allow Policy​

Start both apps with Catalyst Cloud:

diagrid dev run -f dev.yaml --project crewai-access-qs --approve
tip

Wait for both apps to show Uvicorn running on ... in the log output before proceeding.

Open a new terminal and invoke the caller:

curl -X POST http://localhost:8001/call-server \
  -H "Content-Type: application/json" \
  -d '{"topic": "AI in healthcare"}'

You should see a successful 200 response with the crew's research summary. The request traveled from your local caller → Catalyst Cloud → through the tunnel → your local server. By default, Catalyst allows all agent-to-agent communication.

7. Apply a Deny Policy​

Stop the running application by pressing Ctrl+C, then disconnect the app IDs from Catalyst:

diagrid dev stop --app-id caller --project crewai-access-qs
diagrid dev stop --app-id server --project crewai-access-qs

Create a deny-all access configuration for the server:

diagrid configuration create server-deny \
  --project crewai-access-qs \
  --default-action deny

Apply the configuration to the server app ID:

diagrid appid update server \
  --project crewai-access-qs \
  --app-config server-deny

This configuration denies all incoming service invocation requests to the server, regardless of which app is calling it. Because authorization is enforced in Catalyst Cloud — not by the server process — a denied request never reaches your local server. The 403 is returned by Catalyst before any traffic enters the tunnel.

8. Verify Access is Denied​

Restart both apps:

diagrid dev run -f dev.yaml --project crewai-access-qs --approve

Open a new terminal and invoke the caller again:

curl -X POST http://localhost:8001/call-server \
  -H "Content-Type: application/json" \
  -d '{"topic": "AI in healthcare"}'

This time, the caller receives a 403 PermissionDenied error:

{
  "status_code": 403,
  "body": "access control policy has denied access"
}

The access policy blocked the caller from invoking the server. The server's CrewAI crew never executed.

9. Clean Up​

Stop the running application by pressing Ctrl+C.

Delete the Catalyst Cloud project:

diagrid project delete crewai-access-qs

Summary​

In this quickstart, you:

  • Set up two apps communicating through Catalyst service invocation
  • Verified that agent-to-agent calls succeed with the default allow policy
  • Applied a deny-all access configuration that blocked the caller with a 403 PermissionDenied error

Next Steps​