Serve an Operation with HTTP

This guide shows how to expose a CrewMaster operation through a FastAPI HTTP endpoint. Unlike the v1 http_driver approach, you build a standard FastAPI app and call crewmaster.execute() directly — giving you full control over routing, authentication, error handling, and response formats.

Concepts covered

• FastAPI — the web framework
• Operation — typed unit of work
• AgentConfig — persistent agent identity
• ContextStore — domain object registry
• PydanticAIDriver — real LLM driver
• execute() — the execution entry point
• uvicorn — ASGI server

---

What you'll build

A HTTP endpoint that accepts a user query as JSON, runs it through a greeter operation, and returns a typed greeting response. The endpoint is a standard FastAPI route handler — no CrewMaster-specific server machinery.

---

1 Define the request and response models

Separate your HTTP contract from your domain models:

from pydantic import BaseModel, Field


# HTTP contract
class GreetRequest(BaseModel):
    """Incoming request to the greet endpoint."""

    name: str = Field(description="Name of the person to greet")


class GreetResponse(BaseModel):
    """Response from the greet endpoint."""

    message: str = Field(description="The greeting message")
    agent: str = Field(description="Name of the agent that generated it")


# Domain output model (what the Operation produces)
class Greeting(BaseModel):
    """Domain model produced by the greeter operation."""

    message: str

---

2 Set up the agent and operation

The operation setup is identical to the My First Operation tutorial:

from crewmaster.agents.agent import AgentConfig
from crewmaster.operations.operation import Operation

greeter_agent = AgentConfig(
    name="greeter",
    identity_blocks=["blocks://greeter/identity"],
)

greet_operation = Operation(
    name="greet",
    produces=Greeting,
    kind="artifact",
    agent=greeter_agent,
    task_blocks=["blocks://greeter/task"],
)

---

3 Set up prompt blocks

Create the identity and task templates on disk:

from pathlib import Path

from crewmaster.agents.prompts.local_disk_store import LocalDiskStore

BLOCKS_DIR = Path("./blocks")

(BLOCKS_DIR / "greeter" / "identity.j2").mkdir(parents=True, exist_ok=True)
(BLOCKS_DIR / "greeter" / "identity.j2").write_text("""\
---
id: blocks://greeter/identity
kind: identity
provides: greeter_persona
---
You are a warm, friendly greeting assistant. Always address the user
by name and make the greeting personal.
""")

(BLOCKS_DIR / "greeter" / "task.j2").write_text("""\
---
id: blocks://greeter/task
kind: task
provides: greeting_output
---
Write a warm, personal greeting for {{ user_name }}.
""")

block_store = LocalDiskStore(root=str(BLOCKS_DIR))

---

4 Create the FastAPI application

This is a standard FastAPI app. The route handler builds a ContextStore per request, passes context overrides, and calls execute():

import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException

from crewmaster import execute
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.engine import PromptEngine
from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver

# ── Lifespan: create the driver once at startup ───────────────────────
driver: PydanticAIDriver


@asynccontextmanager
async def lifespan(app: FastAPI):
    global driver
    driver = PydanticAIDriver(
        model_name="openai:gpt-4o-mini",
    )
    yield
    # Cleanup (if needed) goes here


app = FastAPI(
    title="CrewMaster Greeter API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.post("/greet", response_model=GreetResponse)
async def greet(request: GreetRequest) -> GreetResponse:
    """Accept a name and return a personalized greeting."""
    # Build a fresh ContextStore per request
    store = ContextStore()

    try:
        result = await execute(
            operation=greet_operation,
            context_store=store,
            default_runtime=driver,
            block_store=block_store,
            prompt_engine=PromptEngine(block_store=block_store),
            context_overrides={"user_name": request.name},
        )
    except Exception as exc:
        raise HTTPException(
            status_code=500,
            detail=f"Operation failed: {exc}",
        ) from exc

    return GreetResponse(
        message=result.message,
        agent=greet_operation.agent.name,
    )

Why create a new ContextStore per request?

Each HTTP request carries its own domain data (the user's name, in this case). A fresh ContextStore ensures no state leaks between requests. If you need to share a read-only dataset across requests (e.g., a product catalog), register it in a store created at startup.

---

5 Run the server

Start the server with uvicorn:

uvicorn serve:app --host 0.0.0.0 --port 8000

Output:

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

---

6 Test the endpoint

Send a POST request with a JSON body:

curl -X POST http://localhost:8000/greet \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice"}'

Expected response:

{
  "message": "Hello, Alice! Welcome aboard!",
  "agent": "greeter"
}

The interactive API documentation is automatically available at http://localhost:8000/docs.

---

7 Complete script

serve.py

"""Serve a CrewMaster operation over HTTP with FastAPI."""
import os
from contextlib import asynccontextmanager
from pathlib import Path

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

from crewmaster import execute
from crewmaster.agents.agent import AgentConfig
from crewmaster.agents.context.store import ContextStore
from crewmaster.agents.prompts.engine import PromptEngine
from crewmaster.agents.prompts.local_disk_store import LocalDiskStore
from crewmaster.execution.drivers.pydantic_ai import PydanticAIDriver
from crewmaster.operations.operation import Operation


# ── Models ──────────────────────────────────────────────────────────
class GreetRequest(BaseModel):
    name: str = Field(description="Name of the person to greet")


class GreetResponse(BaseModel):
    message: str = Field(description="The greeting message")
    agent: str = Field(description="Name of the agent that generated it")


class Greeting(BaseModel):
    message: str


# ── CrewMaster setup ────────────────────────────────────────────────
greeter_agent = AgentConfig(
    name="greeter",
    identity_blocks=["blocks://greeter/identity"],
)

greet_operation = Operation(
    name="greet",
    produces=Greeting,
    kind="artifact",
    agent=greeter_agent,
    task_blocks=["blocks://greeter/task"],
)

BLOCKS_DIR = Path("./blocks")
(BLOCKS_DIR / "greeter" / "identity.j2").mkdir(parents=True, exist_ok=True)
(BLOCKS_DIR / "greeter" / "identity.j2").write_text("""\
---
id: blocks://greeter/identity
kind: identity
provides: greeter_persona
---
You are a warm, friendly greeting assistant. Always address the user
by name and make the greeting personal.
""")

(BLOCKS_DIR / "greeter" / "task.j2").write_text("""\
---
id: blocks://greeter/task
kind: task
provides: greeting_output
---
Write a warm, personal greeting for {{ user_name }}.
""")

block_store = LocalDiskStore(root=str(BLOCKS_DIR))


# ── FastAPI app ─────────────────────────────────────────────────────
driver: PydanticAIDriver


@asynccontextmanager
async def lifespan(app: FastAPI):
    global driver
    driver = PydanticAIDriver(
        model_name="openai:gpt-4o-mini",
    )
    yield


app = FastAPI(
    title="CrewMaster Greeter API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.post("/greet", response_model=GreetResponse)
async def greet(request: GreetRequest) -> GreetResponse:
    store = ContextStore()
    try:
        result = await execute(
            operation=greet_operation,
            context_store=store,
            default_runtime=driver,
            block_store=block_store,
            prompt_engine=PromptEngine(block_store=block_store),
            context_overrides={"user_name": request.name},
        )
    except Exception as exc:
        raise HTTPException(
            status_code=500,
            detail=f"Operation failed: {exc}",
        ) from exc

    return GreetResponse(
        message=result.message,
        agent=greet_operation.agent.name,
    )


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

---

8 Adding tools to the HTTP endpoint

The same pattern extends to tool-equipped operations. Pass a ToolRegistry to execute():

from crewmaster.tools.registry import ToolRegistry
from crewmaster.tools.models import ToolSchema

WEATHER_TOOL = ToolSchema(
    name="get_weather",
    description="Get current weather for a city",
    input_schema={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name"},
        },
        "required": ["city"],
    },
)

tool_registry = ToolRegistry()
tool_registry.register("weather", WEATHER_TOOL)
tool_registry.register("*", WEATHER_TOOL)

# In your route handler:
result = await execute(
    operation=weather_operation,
    context_store=store,
    default_runtime=driver,
    block_store=block_store,
    prompt_engine=PromptEngine(block_store=block_store),
    tool_registry=tool_registry,  # ← pass tools here
    context_overrides={"city": request.city},
)

---

9 Production considerations

Rate limiting and concurrency

LLM APIs have rate limits and can be slow. Consider: - Using a background task queue (Celery, Arq) for long-running operations - Adding request timeouts in your FastAPI app - Caching responses for common queries

Streaming responses

For streaming responses, use execute_stream() and FastAPI's StreamingResponse:

from fastapi.responses import StreamingResponse

@app.post("/greet/stream")
async def greet_stream(request: GreetRequest):
    async def event_generator():
        async for chunk in execute_stream(
            operation=greet_operation,
            context_store=ContextStore(),
            default_runtime=driver,
            block_store=block_store,
        ):
            yield f"data: {chunk.model_dump_json()}\\n\\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

---

Next steps

• Expose a DAG of operations over HTTP with the multi-node pipeline guide.
• Read the Execution API reference to learn about PydanticAIDriver, LangChainDriver, and RuntimeDriver.
• Explore the evaluation with sandbox guide to add evaluation to your HTTP-served operations.