# Quickstart: Zero-Trust Security via Access Policies

Enable authorization for Dapr Agents with access policies.

This quickstart demonstrates how to secure agent-to-agent communication using [Catalyst](https://catalyst.diagrid.io/) access policies. You'll run two apps — a **caller** that invokes a **server** running a Dapr Agents durable agent — and then apply a deny policy to block unauthorized access.

:::note
This tutorial uses [**Catalyst Cloud**](https://docs.diagrid.io/operate/hosting/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](https://docs.diagrid.io/operate/hosting/enterprise-self-hosted) 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

**Diagram: Caller to server through Catalyst service invocation.** Catalyst checks the access policy before it forwards the call. A denied call gets a 403 and never reaches the server.

- **Catalyst**: caller (App ID), Access policy, server (App ID)
- Caller app, Dapr Agents (server)

- Caller app → caller (App ID)
- caller (App ID) → Access policy
- Access policy → server (App ID)
- server (App ID) → Dapr Agents (server)

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 with an 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

- [Diagrid Catalyst account](https://catalyst.diagrid.io/)
- [Diagrid CLI](https://docs.diagrid.io/references/catalyst/catalyst-cli-intro)
- [Python 3.12](https://www.python.org/downloads/)

## 2. Log in to Catalyst

```bash
diagrid login
```

Confirm your identity:

```bash
diagrid whoami
```

## 3. Create project files

Create a new directory for the quickstart:

```bash
mkdir dapr-agents-access-policies
cd dapr-agents-access-policies
mkdir caller
mkdir server
```

### server/app.py

Create `server/app.py` with the following content. It runs a Dapr Agents durable agent behind a `/research` HTTP endpoint. A scripted model client stands in for the LLM: it tells the agent to call the `gather_sources` tool, then `summarize`, so the quickstart needs no LLM API key.

```python
"""Server app: exposes a Dapr Agents durable agent via HTTP."""
import json
import uuid

import uvicorn
from dapr_agents import DurableAgent, tool
from dapr_agents.llm.chat import ChatClientBase
from dapr_agents.types import AssistantMessage, LLMChatCandidate, LLMChatResponse
from dapr_agents.workflow.runners import AgentRunner
from fastapi import FastAPI
from pydantic import BaseModel

# ── Tools ────────────────────────────────────────────────────
@tool
def gather_sources(topic: str) -> str:
    """Gather research sources on a topic."""
    result = f"Found 42 papers on '{topic}'"
    print(f">>> gather_sources: {result}", flush=True)
    return result

@tool
def summarize(topic: str) -> str:
    """Summarize the research on a topic."""
    result = f"Summary: AI diagnostics up 40% for '{topic}'"
    print(f">>> summarize: {result}", flush=True)
    return result

# ── Scripted LLM client (no API key, no network call) ────────
class ScriptedLLM(ChatClientBase):
    """Plays the model's part: call gather_sources, then summarize, then answer."""

    prompty = None
    prompt_template = None

    @classmethod
    def from_prompty(cls, prompty_source, timeout=1500):
        return cls()

    def generate(self, messages=None, response_format=None, **kwargs):
        if response_format is not None:  # end-of-run memory summary
            return response_format(summary="Research complete.")

        def field(m, key):
            return m.get(key) if isinstance(m, dict) else getattr(m, key, None)

        # Only look at the current turn: the last user message and what follows it.
        messages = list(messages or [])
        start = max(i for i, m in enumerate(messages) if field(m, "role") == "user")
        topic = field(messages[start], "content")
        results = [field(m, "content") for m in messages[start:] if field(m, "role") == "tool"]
        steps = ["gather_sources", "summarize"]
        if len(results) < len(steps):
            name = steps[len(results)]
            message = AssistantMessage(content=None, tool_calls=[{
                "id": f"call_{uuid.uuid4().hex}",  # unique across runs
                "type": "function",
                "function": {"name": name, "arguments": json.dumps({"topic": topic})},
            }])
        else:
            message = AssistantMessage(content=json.dumps(results))
        return LLMChatResponse(results=[LLMChatCandidate(message=message)])

# ── Agent ────────────────────────────────────────────────────
agent = DurableAgent(
    name="researcher",
    role="Research Analyst",
    instructions=["Gather sources on the topic, then summarize them."],
    tools=[gather_sources, summarize],
    llm=ScriptedLLM(),
)
runner = AgentRunner()

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

class RunRequest(BaseModel):
    topic: str

@app.post("/research")
async def research(req: RunRequest):
    output = await runner.run(agent, {"task": req.topic})
    return {"messages": json.loads(json.loads(output)["content"])}

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

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

A `DurableAgent` runs as a Dapr workflow, so it needs a workflow state store. `diagrid dev run` enables the Catalyst managed workflow store for the project, so there is nothing to configure. The agent also prints its own conversation trace in the log, next to the `>>>` lines.

### caller/app.py

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

```python
"""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:

**macOS/Linux**

```yaml
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"]
```

**Windows**

```yaml
version: 1
common:
  appLogDestination: console
apps:
  - appID: caller
    appDirPath: ./caller
    command: [".venv\\Scripts\\python.exe", "app.py"]
  - appID: server
    appDirPath: ./server
    appPort: 8002
    command: [".venv\\Scripts\\python.exe", "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. Install dependencies

**macOS/Linux**

```bash
python3 -m venv .venv
source .venv/bin/activate
pip install dapr-agents==1.0.9 fastapi==0.143.0 uvicorn==0.54.0 httpx==0.28.1
```

**Windows**

```powershell
python -m venv .venv
.venv\Scripts\python.exe -m pip install dapr-agents==1.0.9 fastapi==0.143.0 uvicorn==0.54.0 httpx==0.28.1
```

Create symlinks so both apps share the same virtual environment:

**macOS/Linux**

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

**Windows**

```powershell
cmd /c mklink /J caller\.venv .venv
cmd /c mklink /J server\.venv .venv
```

## 5. Run with default allow policy

Start both apps with Catalyst Cloud:

```bash
diagrid dev run -f dev.yaml --project dapr-agents-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:

**macOS/Linux**

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

**Windows**

```powershell
Invoke-RestMethod -Method Post -Uri "http://localhost:8001/call-server" `
  -ContentType "application/json" `
  -Body '{"topic": "AI in healthcare"}'
```

You should see a successful response:

```json
{
  "status_code": 200,
  "body": {
    "messages": [
      "Found 42 papers on 'AI in healthcare'",
      "Summary: AI diagnostics up 40% for 'AI in healthcare'"
    ]
  }
}
```

By default, Catalyst allows all agent-to-agent communication. The request traveled from your local caller → Catalyst Cloud → through the tunnel → your local server. The caller successfully invoked the server's Dapr Agents agent.

## 6. Apply a deny policy

Stop the running application by pressing `Ctrl+C`, then disconnect the apps from Catalyst:

```bash
diagrid dev stop --id caller --project dapr-agents-access-qs
diagrid dev stop --id server --project dapr-agents-access-qs
```

Deny every caller on the server app:

```bash
diagrid app access grant server --project dapr-agents-access-qs --caller '*' --action deny --wait
```

:::tip
If the command reports `local app connection already exists`, wait a few seconds and run it again.
:::

Passing `'*'` as the caller sets the policy's default action instead of naming one app, so this 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.

## 7. Verify access is denied

Restart both apps:

```bash
diagrid dev run -f dev.yaml --project dapr-agents-access-qs --approve
```

Open a **new terminal** and invoke the caller again:

**macOS/Linux**

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

**Windows**

```powershell
Invoke-RestMethod -Method Post -Uri "http://localhost:8001/call-server" `
  -ContentType "application/json" `
  -Body '{"topic": "AI in healthcare"}'
```

This time, the caller receives a **403 PermissionDenied** error:

```json
{
  "status_code": 403,
  "body": "{\"errorCode\":\"ERR_DIRECT_INVOKE\",\"message\":\"failed to invoke, id: server, err: rpc error: code = PermissionDenied desc = access control policy has denied access to id: spiffe://<trust-domain>/ns/<project-id>/caller operation: research verb: POST\"}"
}
```

The access policy blocked the caller from invoking the server. The server's Dapr Agents agent never executed.

## 8. Clean up

Stop the running application by pressing `Ctrl+C`.

Delete the Catalyst Cloud project:

```bash
diagrid project delete dapr-agents-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 policy that blocked the caller with a `403 PermissionDenied` error

## Next steps

- Learn more about [Catalyst Enterprise](https://docs.diagrid.io/operate/hosting/enterprise-self-hosted) for fine-grained authorization rules
- Explore the [Catalyst web console](https://catalyst.diagrid.io/) for managing configurations and apps