CrewAI Tutorial (Python): adding authentication for intermediate developers

This tutorial shows how to add authentication to a CrewAI-based Python workflow so only valid users or services can trigger agent execution. You need this when your crew is exposed through an API, internal tool, or automation layer and you want to stop anonymous access before any LLM calls happen.

What You'll Need

• •Python 3.10+
• •A CrewAI project installed with:
  • •crewai
  • •crewai-tools
  • •fastapi
  • •uvicorn
  • •python-dotenv
• •An LLM API key, such as:
  • •OPENAI_API_KEY
• •A shared secret for request authentication:
  • •API_AUTH_TOKEN
• •Basic familiarity with:
  • •CrewAI agents, tasks, and crews
  • •FastAPI request handling

Step-by-Step

1. •Create a minimal CrewAI crew first.
   Keep the crew simple so the auth layer stays isolated and easy to test.

from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

llm = LLM(model="gpt-4o-mini")

researcher = Agent(
    role="Researcher",
    goal="Summarize a topic clearly",
    backstory="You are a concise research assistant.",
    llm=llm,
)

task = Task(
    description="Explain what JWT authentication is in one paragraph.",
    expected_output="A short explanation of JWT authentication.",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[task],
    process=Process.sequential,
)

result = crew.kickoff()
print(result)

2. •Add configuration for secrets and load them from environment variables.
   Do not hardcode tokens in source code; keep auth material outside the repo.

import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
API_AUTH_TOKEN = os.getenv("API_AUTH_TOKEN")

if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY is required")

if not API_AUTH_TOKEN:
    raise ValueError("API_AUTH_TOKEN is required")

3. •Wrap the crew behind a FastAPI endpoint that checks the bearer token before execution.
   This is the main control point: if the token is wrong, the request never reaches CrewAI.

from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel

app = FastAPI()

class RunRequest(BaseModel):
    topic: str

def verify_token(authorization: str | None) -> None:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")

    token = authorization.removeprefix("Bearer ").strip()
    if token != API_AUTH_TOKEN:
        raise HTTPException(status_code=403, detail="Invalid token")

@app.post("/run")
def run_crew(payload: RunRequest, authorization: str | None = Header(default=None)):
    verify_token(authorization)

    task.description = f"Explain {payload.topic} in one paragraph."
    result = crew.kickoff()
    return {"output": str(result)}

4. •Put the app and crew setup into one runnable file.
   This version is executable as-is and keeps the auth logic close to the API boundary.

import os
from dotenv import load_dotenv
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process
from crewai.llm import LLM

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
API_AUTH_TOKEN = os.getenv("API_AUTH_TOKEN")

if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY is required")
if not API_AUTH_TOKEN:
    raise ValueError("API_AUTH_TOKEN is required")

llm = LLM(model="gpt-4o-mini", api_key=OPENAI_API_KEY)
app = FastAPI()

researcher = Agent(
    role="Researcher",
    goal="Summarize a topic clearly",
    backstory="You are a concise research assistant.",
    llm=llm,
)

class RunRequest(BaseModel):
    topic: str

def build_crew(topic: str) -> Crew:
    task = Task(
        description=f"Explain {topic} in one paragraph.",
        expected_output="A short explanation of the topic.",
        agent=researcher,
    )
    return Crew(agents=[researcher], tasks=[task], process=Process.sequential)

def verify_token(authorization: str | None) -> None:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing bearer token")
    token = authorization.removeprefix("Bearer ").strip()
    if token != API_AUTH_TOKEN:
        raise HTTPException(status_code=403, detail="Invalid token")

@app.post("/run")
def run_crew(payload: RunRequest, authorization: str | None = Header(default=None)):
    verify_token(authorization)
    crew = build_crew(payload.topic)
    result = crew.kickoff()
    return {"output": str(result)}

5. •Run the service and call it with and without a valid token.
   This confirms both sides of the behavior: unauthorized requests fail fast, authorized requests reach the crew.

export OPENAI_API_KEY="your-openai-key"
export API_AUTH_TOKEN="super-secret-token"
uvicorn app:app --reload --port 8000

Testing It

Start by sending a request without an Authorization header; you should get a 401 Missing bearer token. Then send one with the wrong bearer token and confirm it returns 403 Invalid token. Finally, send a request with Authorization: Bearer super-secret-token and verify that you get a normal CrewAI response back from /run.

Use curl for a quick check:

curl -X POST "http://127.0.0.1:8000/run" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer super-secret-token" \
  -d '{"topic":"JWT authentication"}'

If you want stronger verification, add integration tests with FastAPI’s TestClient and assert that unauthorized requests never instantiate or run your crew logic.

Next Steps

• •Replace the shared secret with JWT validation using pyjwt and a public key or issuer config.
• •Move auth into FastAPI dependencies so multiple endpoints can reuse the same guard.
• •Add role-based access control if different users should trigger different crews or task sets.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.