Skip to main content
Task Apps are simple, lightweight FastAPI services that wrap your environment logic for training. Everything about the contract—endpoints, schemas, tracing hooks—already exists in the SDK; you only implement the task-specific bits inside your rollout executor. In practice you create one Python file (per task) that wires everything together: dataset loaders, rollout executor, TaskAppConfig, and a small FastAPI entrypoint or registry call. The sections below walk through the expected order so you always know where each piece goes.

​
Anatomy of a task-app file

  1. Imports & constants: Define dataset specs, reward constants, and helper functions you’ll reuse.
  2. Dataset registry: Build a TaskDatasetRegistry, register splits, and expose helpers such as build_dataset() so everything is cached.
  3. Task description helpers: Provide _base_task_info() plus any describe_taskset/provide_task_instances helpers.
  4. Rollout executor: Implement async def rollout_executor(...) that runs your environment and returns a RolloutResponse.
  5. TaskAppConfig builder: Create build_config() that stitches dataset, TaskInfo helpers, rollout executor, tracing hooks, and proxies together.
  6. Registration & entrypoints: Call register_task_app(...) for discovery and optionally expose a fastapi_app()/create_app() function or run_task_app(...) block for local serving.

​
Build the TaskAppConfig

This section shows the build_config() function you add near the end of your file. It pulls together the dataset, task metadata, rollout executor, tracing hooks, SFT output path, proxy settings, and routers—everything the FastAPI harness needs to serve your task. Copy the structure, then replace the dataset/loaders/rubrics with your own. 
from synth_ai.task.server import ProxyConfig, RubricBundle, TaskAppConfig
from synth_ai.task.tracing_utils import (
    build_tracer_factory,
    resolve_sft_output_dir,
    resolve_tracing_db_url,
    tracing_env_enabled,
)
from synth_ai.tracing_v3.session_tracer import SessionTracer

def build_config() -> TaskAppConfig:
    registry, dataset = build_dataset()
    base_info = _base_task_info()

    tracing_enabled = tracing_env_enabled()
    tracer_factory = build_tracer_factory(
        SessionTracer,
        enabled=tracing_enabled,
        db_url=resolve_tracing_db_url(),
    )
    sft_output_dir = resolve_sft_output_dir()

    app_state = {
        "math_dataset": dataset,
        "tracing_enabled": tracing_enabled,
    }
    if tracer_factory:
        app_state["session_tracer_factory"] = tracer_factory
    if sft_output_dir:
        app_state["sft_output_dir"] = sft_output_dir

    proxy = ProxyConfig(enable_openai=True, enable_groq=True)

    return TaskAppConfig(
        app_id="math-single-step",
        name="Math Single Step Task",
        base_task_info=base_info,
        describe_taskset=lambda: describe_taskset(dataset),
        provide_task_instances=lambda seeds: provide_task_instances(dataset, seeds),
        rollout=rollout_executor,
        dataset_registry=registry,
        rubrics=RubricBundle(outcome=OUTCOME_RUBRIC, events=EVENTS_RUBRIC),
        proxy=proxy,
        routers=(math_router,),
        app_state=app_state,
        cors_origins=["*"],
    )
A few points to note:
  • build_dataset() warms the dataset and registers it once, so both RL and SFT share the same loader cache.
  • app_state is how you pass handles (dataset managers, tracer factories, SFT destinations) to routers and rollout code.
  • register_task_app makes the config discoverable by the CLI and deployment tooling.

​
Routes and FastAPI entrypoints

create_task_app turns your TaskAppConfig into an actual FastAPI app and auto-mounts /, /health, /info, /task_info, /rollout, /done, /debug/env, and any proxy routes. Most teams expose that ASGI app through a thin wrapper so local dev, demos, and deployments all hit the same routes.
  • / confirms the service is alive and names the task: 
    @app.get("/")
async def root():
    return {"status": "ok", "service": cfg.app_id}
  • /health enforces the environment API key and returns the expected prefix plus auth metadata: 
    @app.get("/health")
async def health(request: Request):
    expected = normalize_environment_api_key()
    return {
        "healthy": True,
        "auth": {
            "required": True,
            "expected_prefix": (expected[:6] + "...") if expected else "<unset>",
        },
    }
  • /info surfaces dataset, rubric, inference, and limit metadata pulled from your TaskAppConfig.base_task_info. 
    @app.get("/info")
async def info():
    return {
        "service": {"task": cfg.base_task_info.task, "version": cfg.base_task_info.task.version},
        "dataset": cfg.base_task_info.dataset,
        "rubrics": ...,
        "inference": cfg.base_task_info.inference,
        "limits": cfg.base_task_info.limits,
    }
  • /task_info returns either the taskset descriptor or concrete TaskInfo instances for the requested seeds. 
    @app.get("/task_info")
async def task_info(seed: Sequence[int] | None = Query(None), seeds: Sequence[int] | None = Query(None)):
    if not any([seed, seeds]):
        return {"taskset": await cfg.describe_taskset()}
    instances = await cfg.provide_task_instances(all_seeds)
    return [TaskInfo.model_validate(instance).model_dump() for instance in instances]
  • /rollout accepts a RolloutRequest, calls your executor, and normalizes the response into a RolloutResponse. 
    @app.post("/rollout")
async def rollout_endpoint(rollout_request: RolloutRequest, request: Request):
    result = await cfg.rollout(rollout_request, request)
    return RolloutResponse.model_validate(result).model_dump()
  • /done is a lightweight readiness hook used by automation, and /debug/env exposes masked API-key info for debugging missing secrets.
If you need to override specific routes, follow the math legacy task app: it filters the /health routes installed by create_task_app and registers bespoke handlers that surface API-key expectations plus structured 422 logs. The pattern is: 
app = create_task_app(build_config())
app.router.routes = [r for r in app.router.routes if getattr(r, "path", None) not in {"/health", "/health/rollout"}]

@app.get("/health")
async def health(request: Request):
    ...

@app.get("/health/rollout")
async def health_rollout(request: Request):
    ...

​
What’s next

Validate your task app is ready to deploy by running 
uvx synth-ai task-app check ./task_app.py # <- path to your task app
Once your task app is verified: