Chapter 22 · Deployment and Service

Chapter goals: Compare CrewAI Enterprise managed hosting vs self-hosted routes, wrap a Crew as an async task API with FastAPI, master environment variable injection, long-task checkpoint recovery, and stateless horizontal scaling.

22.1 Choosing Between Two Deployment Routes

Dimension	CrewAI Enterprise (Managed)	Self-Hosted (this section's focus)
Setup cost	crewai deploy push-to-production	Write your own service layer; manage processes
Runtime environment	Platform-managed; auto-scales	Docker/K8s self-managed
Data boundary	Runs on platform side	Fully self-owned
Flexibility	Constrained by platform	Arbitrary modification

Learning path recommendation

Get self-hosted running first — go through "API-ification → timeout recovery → scaling" — so you understand all operational concerns before evaluating migration to Enterprise to save ops overhead. The concepts map one-to-one: deployment = service-ification, trigger = kickoff, logs = the observability system from Chapter 21.

The minimal self-hosted形态 is a Web service that accepts input, executes crew.kickoff(), and returns the result.

22.2 Wrapping kickoff with FastAPI: Async Task + Polling

In production a single kickoff can run for minutes; synchronous HTTP waiting will time out. The standard pattern is POST submit task returns job_id → background execution → GET poll for status:

# main.py
import os
import uuid
from fastapi import FastAPI, HTTPException, BackgroundTasks

from marketing_crew import MarketingCrew  # Your crew definition (Chapter 23)

app = FastAPI(title="Marketing Crew API")

# In-process job table; swap to Redis/database for multi-replica deployment (see 22.4)
JOBS: dict[str, dict] = {}


def run_crew_job(job_id: str, topic: str) -> None:
    """Execute crew synchronously in a thread pool and write result back to job table"""
    try:
        result = MarketingCrew().crew().kickoff(inputs={"topic": topic})
        JOBS[job_id].update(
            status="completed",
            result=result.raw,
            tokens=result.token_usage.total_tokens,
        )
    except Exception as e:
        JOBS[job_id].update(status="failed", error=str(e))


@app.post("/kickoff")
def kickoff(topic: str, background_tasks: BackgroundTasks):
    """Submit a crew run; immediately return job_id"""
    if not os.getenv("DEEPSEEK_API_KEY"):
        raise HTTPException(500, "Missing DEEPSEEK_API_KEY environment variable")
    job_id = uuid.uuid4().hex
    JOBS[job_id] = {"status": "running", "result": None}
    # FastAPI BackgroundTasks runs the function after the response is returned
    background_tasks.add_task(run_crew_job, job_id, topic)
    return {"job_id": job_id, "status": "running"}


@app.get("/status/{job_id}")
def status(job_id: str):
    """Client polls task status"""
    job = JOBS.get(job_id)
    if not job:
        raise HTTPException(404, "Job not found")
    return job


# Start: uvicorn main:app --host 0.0.0.0 --port 8000
# curl -X POST "http://localhost:8000/kickoff?topic=AI+Agent"
# curl http://localhost:8000/status/<job_id>

Key points: BackgroundTasks makes HTTP return immediately; the Crew itself is sync-blocking, so it goes into a background thread rather than being called directly in async def (which would deadlock the event loop). For stronger concurrency control, switch to asyncio.to_thread or a standalone task queue (Celery/RQ/Arq).

22.3 Credentials and Environment Variable Injection

Never hardcode API keys into code or images under any circumstances:

# Dockerfile — multi-stage build; credentials injected only at runtime
FROM python:3.12-slim AS base
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# No secrets in the image!
ENV PYTHONUNBUFFERED=1
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# Inject via env file at runtime (.env added to .gitignore)
docker run --rm -p 8000:8000 \
  --env-file .env \
  my-crew-api:latest

# .env sample:
# DEEPSEEK_API_KEY=sk-xxx
# SERPER_API_KEY=xxx

CrewAI reads *_API_KEY and similar variables from the process environment; in Kubernetes scenarios mount Secrets as environment variables — the effect is identical.

22.4 Long-Task Timeouts and Checkpoint Recovery

What if the process is killed mid-run (OOM, deploy restart, upstream LLM jitter)? CrewAI has a built-in checkpoint mechanism: saves recoverable full-state snapshots per-event; on恢复, auto-skips already-completed tasks:

from crewai import Agent, Task, Crew

researcher = Agent(role="Researcher", goal="Research industry trends", backstory="Senior analyst")
writer = Agent(role="Writer", goal="Write report", backstory="Technical writer")

crew = Crew(
    agents=[researcher, writer],
    tasks=[
        Task(description="Research AI Agent market", agent=researcher,
             expected_output="Key point list"),
        Task(description="Write summary report", agent=writer,
             expected_output="500-word report"),
    ],
    checkpoint=True,  # Enable event-driven checkpoints; default: one snapshot per task_completed to ./.checkpoints/
)

result = crew.kickoff()  # After mid-run Ctrl+C or process kill:
# ./.checkpoints/<timestamp>_<uuid>.json is the snapshot

from pathlib import Path

# Resume from the latest snapshot: completed tasks are skipped; resume the remainder
checkpoint_file = sorted(Path("./.checkpoints").glob("*.json"))[-1]
resumed = Crew(
    agents=[researcher, writer],
    tasks=crew.tasks,
    checkpoint=checkpoint_file,   # Pass snapshot file to enter resume mode
)
print(resumed.kickoff().raw)

配套参数: storage providers include JsonProvider (one file per snapshot, easy to inspect) and SqliteProvider (single DB for high-frequency writes); max_checkpoints controls retention count, auto-evicting the oldest snapshots. Note auto-checkpoints are best-effort — write failure logs but does not interrupt execution; manual state.checkpoint() raises on failure. Mount .checkpoints/ as a persistent卷; paired with the job table from 22.2, this enables "resume unfinished tasks after service restart."

22.5 Horizontal Scaling: Making the Service Stateless

Before deploying multiple replicas, run a stateless audit. Four must-check items:

1. Externalize the job table: the 22.2 JOBS dict is process-memory state — across replicas, polling lands on the instance that doesn't have that task. Swap to Redis (SET job:<id> ... EX 3600) or a database table;
2. File artifacts to object storage: output_file writes to local disk are invisible across replicas; switch to S3/OSS;
3. Memory/knowledge base externalized: when enabling memory/knowledge, configure a shared backend (e.g., external embedder and store) to avoid each replica storing its own copy;
4. Checkpoint directory shared: mount .checkpoints/ as a network volume or switch to SqliteProvider on a shared DB; otherwise recovery can only land on the original replica.

Scaling itself is simple — Crew execution is CPU+IO-bound processes; uvicorn --workers N or K8s HPA scales by CPU/queue length. The real bottleneck is usually the upstream LLM's rate limits (protect with max_rpm, see Chapter 25).

Redis reference implementation for externalized job table:

# jobs.py — replace in-process dict with Redis; supports multi-replica deployment
import os
import redis

r = redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
TTL = 3600  # Keep job records for 1 hour


def create_job(job_id: str) -> None:
    r.hset(f"job:{job_id}", mapping={"status": "running", "result": ""})
    r.expire(f"job:{job_id}", TTL)


def finish_job(job_id: str, result: str, tokens: int) -> None:
    r.hset(f"job:{job_id}", mapping={
        "status": "completed", "result": result, "tokens": tokens})


def get_job(job_id: str) -> dict | None:
    data = r.hgetall(f"job:{job_id}")
    if not data:
        return None
    return {k.decode(): v.decode() for k, v in data.items()}

Replace all reads/writes to the JOBS dict in 22.2 with these three functions, and any replica's /status can查询 the same job — this is the core step toward statelessness.

Chapter Summary

• Two deployment routes: Enterprise managed saves ops overhead; self-hosted is flexible and controllable — master self-hosted first;
• API-ification standard posture: POST /kickoff returns job_id + BackgroundTasks for background execution + GET /status for polling;
• Credentials injected only at runtime via environment variables; zero secrets in the image build;
• checkpoint=True event-driven snapshot storage; auto-skips completed tasks on resume; JsonProvider easy to inspect, SqliteProvider handles high frequency;
• Stateless four audits: externalize job table, file artifacts, memory/knowledge, checkpoint directory — only then can you scale horizontally.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Why should the /kickoff endpoint use "background task + job_id polling" instead of synchronous wait?

2. Regarding CrewAI checkpoints, which statement is incorrect?

3. After scaling the 22.2 service to 3 replicas, GET /status frequently returns 404. Most likely cause?

4. Regarding credential management in containers, what is the correct approach?

🛠️ Hands-On Exercises

1. Add a DELETE /jobs/{job_id} cancellation endpoint to the 22.2 service: mark the task as cancelled, and ignore late result writes after crew execution ends.
2. Enable checkpoint=True for your Crew; during the second task's execution, manually kill -9 the process; then resume from the latest snapshot and verify task 1 is indeed skipped (confirm via callback log).
3. Refactor the JOBS dict to a Redis implementation (redis-py get/set + TTL); bring up 2 API replicas + 1 Redis with docker compose; verify any replica can query the same job status.

All principles are in place. Next,串 them together with two complete projects: Chapter 23 · Capstone Project One.