Embedded

Run task app, tunnel, and job from a single script

Run task app + tunnel + job from a single Python script. See Task Apps for contracts and rollout handling.

Basic Usage

from synth_ai.sdk.task import InProcessTaskApp

async with InProcessTaskApp(
    task_app_path="my_task_app.py",
    port=8114,
) as task_app:
    print(f"Running at: {task_app.url}")
    # task_app.url is the public tunnel URL

With Job Submission

from synth_ai.sdk.task import InProcessTaskApp
from synth_ai.sdk.training import PromptLearningJob

async with InProcessTaskApp(task_app_path="my_app.py") as task_app:
    job = PromptLearningJob.from_config(
        "gepa.toml",
        overrides={"prompt_learning.task_app_url": task_app.url}
    )
    job.submit()
    result = job.poll_until_complete()
    print(result["best_prompt"])
# Cleaned up automatically

Input Methods

Exactly one required:

Parameter	Type	Description
task_app_path	Path	.py file with build_config() or fastapi_app()
app	FastAPI	Direct app instance
config	TaskAppConfig	Config object
config_factory	Callable	Function returning TaskAppConfig

TaskAppConfig

For complex task apps:

from synth_ai.sdk.task import TaskAppConfig, create_task_app
from synth_ai.sdk.task.contracts import TaskInfo, RolloutRequest, RolloutResponse, RolloutMetrics

def my_rollout(request: RolloutRequest, http_request) -> RolloutResponse:
    seed = request.env.seed or 0
    prompt = request.policy.config.get("prompt_template", {})
    reward = evaluate(prompt, seed)
    
    return RolloutResponse(
        run_id=request.run_id,
        metrics=RolloutMetrics(
            mean_return=reward,
            episode_returns=[reward],
            num_steps=1,
            num_episodes=1,
        ),
        trajectories=[],
    )

def build_config() -> TaskAppConfig:
    return TaskAppConfig(
        app_id="my-classifier",
        name="Intent Classifier",
        description="Classifies intents",
        base_task_info=TaskInfo(...),
        rollout=my_rollout,
        require_api_key=True,
    )

def fastapi_app():
    return create_task_app(build_config())

Key Config Fields

Field	Description
app_id	Unique identifier
name	Human-readable name
rollout	Function handling /rollout
require_api_key	Enforce ENVIRONMENT_API_KEY

All Parameters

Parameter	Type	Default	Description
task_app_path	Path	-	.py file with build_config() or fastapi_app()
app	FastAPI	-	Direct app instance
config	TaskAppConfig	-	Config object
config_factory	Callable	-	Function returning TaskAppConfig
port	int	8114	Local port
host	str	127.0.0.1	Host to bind
tunnel_mode	str	"quick"	Tunnel mode
api_key	str	env	API key for health checks
health_check_timeout	float	30.0	Health check timeout (seconds)
auto_find_port	bool	True	Auto-find available port

Environment

Variable	Purpose
SYNTH_TUNNEL_MODE	"quick" or "local" (skip tunnel)
SYNTH_TUNNEL_HOSTNAME	Override tunnel hostname
ENVIRONMENT_API_KEY	Auth key for /rollout

CLI Alternative

uvx synth-ai deploy tunnel my_app.py --tunnel-mode quick --env .env