Deploy CrewAI as a FastAPI Service on a VPS

apt update && apt upgrade -y
apt install -y curl wget git build-essential ca-certificates gnupg ufw fail2ban
timedatectl set-timezone UTC

adduser admin && usermod -aG sudo admin
mkdir -p /home/admin/.ssh && cp /root/.ssh/authorized_keys /home/admin/.ssh/
chown -R admin:admin /home/admin/.ssh && chmod 700 /home/admin/.ssh
chmod 600 /home/admin/.ssh/authorized_keys

adduser --system --group --home /opt/crewai --shell /bin/bash crewai

# /etc/ssh/sshd_config.d/99-hardening.conf
PermitRootLogin no
PasswordAuthentication no
KbdInteractiveAuthentication no

systemctl reload ssh
ufw default deny incoming && ufw default allow outgoing
ufw allow 22/tcp && ufw enable

sudo apt install -y python3-dev python3-venv python3-pip libssl-dev libffi-dev
curl -LsSf https://astral.sh/uv/install.sh | sudo UV_INSTALL_DIR=/usr/local/bin sh
uv --version

sudo -iu crewai
uv tool install crewai
crewai create crew researcher_api
cd researcher_api
uv sync

export OPENAI_API_KEY="sk-your-key-here"
uv run run_crew  # smoke test

uv add fastapi "uvicorn[standard]" gunicorn pydantic-settings

from __future__ import annotations
import logging, uuid
from datetime import datetime, timezone
from typing import Any
from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException, status
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings, SettingsConfigDict
from researcher_api.crew import ResearcherApi

logger = logging.getLogger("crewai.api")

class Settings(BaseSettings):
    api_token: str = Field(min_length=24)
    app_env: str = "production"
    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()

class JobRequest(BaseModel):
    topic: str = Field(min_length=2, max_length=200)
    extra_inputs: dict[str, Any] = Field(default_factory=dict)

class JobStatus(BaseModel):
    job_id: str; status: str; created_at: datetime
    finished_at: datetime | None = None
    result: Any | None = None; error: str | None = None

JOBS: dict[str, JobStatus] = {}

def require_token(x_api_token: str = Header(default="")) -> None:
    if x_api_token != settings.api_token:
        raise HTTPException(status_code=401, detail="Invalid token")

app = FastAPI(title="CrewAI Researcher API",
    docs_url="/docs" if settings.app_env != "production" else None)

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

def _run_crew(job_id: str, inputs: dict[str, Any]) -> None:
    JOBS[job_id].status = "running"
    try:
        result = ResearcherApi().crew().kickoff(inputs=inputs)
        JOBS[job_id].result = str(result); JOBS[job_id].status = "succeeded"
    except Exception as exc:
        logger.exception("Crew job %s failed", job_id)
        JOBS[job_id].error = str(exc); JOBS[job_id].status = "failed"
    finally:
        JOBS[job_id].finished_at = datetime.now(timezone.utc)

@app.post("/jobs", response_model=JobStatus, status_code=202,
    dependencies=[Depends(require_token)])
def create_job(req: JobRequest, background: BackgroundTasks) -> JobStatus:
    job_id = uuid.uuid4().hex
    job = JobStatus(job_id=job_id, status="queued", created_at=datetime.now(timezone.utc))
    JOBS[job_id] = job
    background.add_task(_run_crew, job_id, {"topic": req.topic, **req.extra_inputs})
    return job

@app.get("/jobs/{job_id}", response_model=JobStatus,
    dependencies=[Depends(require_token)])
def get_job(job_id: str) -> JobStatus:
    job = JOBS.get(job_id)
    if not job: raise HTTPException(404, "Not found")
    return job

python3 -c "import secrets; print(secrets.token_urlsafe(48))"

APP_ENV=production
API_TOKEN=<paste-token>
OPENAI_API_KEY=sk-your-key-here
# Or any other LiteLLM-supported provider

chmod 600 .env

import multiprocessing
bind = "127.0.0.1:8000"
workers = max(2, multiprocessing.cpu_count())
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 600
graceful_timeout = 30
keepalive = 5
accesslog = "/var/log/crewai/access.log"
errorlog = "/var/log/crewai/error.log"
loglevel = "info"

# /etc/systemd/system/crewai-api.service
[Unit]
Description=CrewAI FastAPI service
After=network.target

[Service]
Type=simple
User=crewai
Group=crewai
WorkingDirectory=/opt/crewai/researcher_api
EnvironmentFile=/opt/crewai/researcher_api/.env
ExecStart=/home/crewai/.local/bin/uv run gunicorn -c gunicorn.conf.py researcher_api.api:app
Restart=on-failure
RestartSec=5

NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/opt/crewai /var/log/crewai
ProtectKernelTunables=true
ProtectKernelModules=true

[Install]
WantedBy=multi-user.target

sudo mkdir -p /var/log/crewai && sudo chown crewai:crewai /var/log/crewai
sudo systemctl daemon-reload
sudo systemctl enable --now crewai-api
sudo systemctl status crewai-api

sudo apt install -y nginx certbot python3-certbot-nginx
sudo ufw allow 80/tcp && sudo ufw allow 443/tcp

server {
  listen 80;
  server_name api.example.com;
  return 301 https://$host$request_uri;
}
server {
  listen 443 ssl http2;
  server_name api.example.com;
  ssl_protocols TLSv1.2 TLSv1.3;
  add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
  client_max_body_size 1m;

  location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header Host $host;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_read_timeout 600s;
    proxy_send_timeout 600s;
  }
}

sudo ln -s /etc/nginx/sites-available/crewai-api.conf /etc/nginx/sites-enabled/
sudo rm -f /etc/nginx/sites-enabled/default
sudo nginx -t && sudo systemctl reload nginx
sudo certbot --nginx -d api.example.com --redirect --agree-tos -m admin@example.com -n

limit_req_zone $binary_remote_addr zone=crewai_api:10m rate=10r/s;
limit_req_status 429;

# Inside server block, location /:
limit_req zone=crewai_api burst=20 nodelay;

[sshd]
enabled = true
maxretry = 4
bantime  = 1h

[nginx-limit-req]
enabled  = true
port     = http,https
filter   = nginx-limit-req
logpath  = /var/log/nginx/error.log
maxretry = 10
bantime  = 1h

/var/log/crewai/*.log {
  daily
  rotate 14
  missingok
  notifempty
  compress
  delaycompress
  create 0640 crewai crewai
  sharedscripts
  postrotate
    systemctl kill -s USR1 crewai-api.service > /dev/null 2>&1 || true
  endscript
}

uv add prometheus-fastapi-instrumentator

# in api.py after app = FastAPI(...):
from prometheus_fastapi_instrumentator import Instrumentator
Instrumentator().instrument(app).expose(app, endpoint="/metrics", include_in_schema=False)