CrewAI Tutorial (Python): adding authentication for beginners

This tutorial shows you how to add authentication to a CrewAI-based Python app so only approved users can trigger agents and access results. You need this when your crew is exposed through an API, internal tool, or admin panel and you want a simple gate before any agent work starts.

What You'll Need

• •Python 3.10 or newer
• •A CrewAI project already installed
• •crewai package
• •fastapi and uvicorn for the API layer
• •python-dotenv for local secrets
• •An API key for your LLM provider, such as OPENAI_API_KEY
• •A shared authentication secret, such as AUTH_TOKEN

Install the dependencies:

pip install crewai fastapi uvicorn python-dotenv openai

Step-by-Step

1. •Start with a minimal CrewAI setup.
   The crew itself does not handle authentication, so keep that logic outside the agent layer. Your agents should only run after the request has already been verified.

from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Researcher",
    goal="Find concise answers to user questions",
    backstory="You are careful and factual.",
    verbose=False,
)

task = Task(
    description="Answer the user's question clearly and briefly.",
    expected_output="A short answer with useful details.",
    agent=researcher,
)

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

2. •Load secrets from environment variables.
   This keeps your auth token and LLM key out of source control. For beginners, a single bearer token is enough to protect an internal demo or prototype.

import os
from dotenv import load_dotenv

load_dotenv()

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

if not OPENAI_API_KEY:
    raise RuntimeError("Missing OPENAI_API_KEY")
if not AUTH_TOKEN:
    raise RuntimeError("Missing AUTH_TOKEN")

3. •Add a small authentication helper.
   This function checks an incoming bearer token against your shared secret. In production you would replace this with JWT validation, OAuth2, or your company identity provider.

from fastapi import Header, HTTPException

def require_auth(authorization: str | None):
    if not authorization:
        raise HTTPException(status_code=401, detail="Missing Authorization header")

    prefix = "Bearer "
    if not authorization.startswith(prefix):
        raise HTTPException(status_code=401, detail="Invalid auth scheme")

    token = authorization[len(prefix):].strip()
    if token != AUTH_TOKEN:
        raise HTTPException(status_code=403, detail="Invalid token")

    return True

4. •Wrap the crew in a FastAPI endpoint.
   The endpoint verifies the request before calling crew.kickoff(). This is the important part: never let unauthenticated traffic reach your agent execution path.

from fastapi import FastAPI, Header

app = FastAPI()

@app.post("/ask")
def ask(question: str, authorization: str | None = Header(default=None)):
    require_auth(authorization)

    result = crew.kickoff(inputs={"question": question})
    return {"answer": str(result)}

5. •Put everything together in one runnable file.
   This version uses OpenAI through CrewAI and exposes a protected /ask endpoint. Save it as main.py.

import os
from dotenv import load_dotenv
from fastapi import FastAPI, Header, HTTPException
from crewai import Agent, Task, Crew, Process

load_dotenv()

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

if not OPENAI_API_KEY:
    raise RuntimeError("Missing OPENAI_API_KEY")
if not AUTH_TOKEN:
    raise RuntimeError("Missing AUTH_TOKEN")

os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY

researcher = Agent(
    role="Researcher",
    goal="Find concise answers to user questions",
    backstory="You are careful and factual.",
)

task = Task(
    description="Answer the user's question: {question}",
    expected_output="A short answer with useful details.",
    agent=researcher,
)

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

def require_auth(authorization: str | None):
    if not authorization:
        raise HTTPException(status_code=401, detail="Missing Authorization header")
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid auth scheme")
    if authorization.removeprefix("Bearer ").strip() != AUTH_TOKEN:
        raise HTTPException(status_code=403, detail="Invalid token")

@app.post("/ask")
def ask(question: str, authorization: str | None = Header(default=None)):
    require_auth(authorization)
    result = crew.kickoff(inputs={"question": question})
    return {"answer": str(result)}

6. •Run the API locally with Uvicorn.
   Use one terminal for the server and another for testing requests. Keep the .env file private.

export OPENAI_API_KEY="your-openai-key"
export AUTH_TOKEN="super-secret-token"
uvicorn main:app --reload

Testing It

First test without credentials; you should get a 401 Missing Authorization header. Then send a bad token and confirm you get 403 Invalid token. Finally send Authorization: Bearer super-secret-token and verify that the crew runs and returns an answer.

Example request:

curl -X POST "http://127.0.0.1:8000/ask?question=What%20is%20CrewAI%3F" \
  -H "Authorization: Bearer super-secret-token"

If that works, your auth gate is correctly protecting the agent call path.

Next Steps

• •Replace the shared bearer token with JWT validation using your identity provider.
• •Add role-based access control so different users can trigger different crews or tasks.
• •Move authentication into middleware once you have multiple protected endpoints instead of one route.

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.