In-Process Task App

Run task apps entirely within your Python script with automatic tunnel management

The InProcessTaskApp utility enables running task apps entirely within your Python script, eliminating the need for separate terminal processes or manual tunnel management. It automatically starts a FastAPI server, opens a tunnel (SynthTunnel by default, Cloudflare optional), and provides the tunnel URL for GEPA optimization jobs.

In-process task apps are designed for production workflows that run GEPA or eval on the fly, where you want the task app lifecycle to be managed inside the same service or worker that triggers the job.

Quick Start

from synth_ai.sdk import InProcessTaskApp
from your_task_app import build_config

async def main():
    async with InProcessTaskApp(
        config_factory=build_config,
        port=8114,
    ) as task_app:
        print(f"Task app running at: {task_app.url}")
        # Use task_app.url for your optimization jobs
        # The tunnel and server will automatically clean up on exit

asyncio.run(main())

Features

Automatic Server Management

Starts uvicorn server in background thread

Automatic Tunnel Creation

Opens SynthTunnel by default (Cloudflare optional)

Port Conflict Handling

Automatically finds available ports if requested port is busy

Signal Handling

Graceful shutdown on SIGINT/SIGTERM

Observability

Structured logging and optional callbacks

Multiple Input Methods

Supports app, config, config_factory, or file path

API Reference

InProcessTaskApp

Context manager for running task apps in-process with automatic tunneling.

Parameters

Parameter	Type	Default	Description
app	Optional[ASGIApplication]	None	FastAPI app instance (most direct)
config	Optional[TaskAppConfig]	None	TaskAppConfig object
config_factory	Optional[Callable[[], TaskAppConfig]]	None	Callable that returns TaskAppConfig
task_app_path	Optional[Path | str]	None	Path to task app .py file (fallback)
port	int	8114	Local port to run server on (must be in range [1024, 65535])
host	str	"127.0.0.1"	Host to bind to (must be localhost for security)
tunnel_mode	str	"synthtunnel"	Tunnel mode (“synthtunnel”, “quick”, “named”, “local”, “preconfigured”)
tunnel_backend	Optional[TunnelBackend | str]	None	Explicit tunnel backend (overrides tunnel_mode)
preconfigured_url	Optional[str]	None	External URL when tunnel_mode="preconfigured"
preconfigured_auth_header	Optional[str]	None	Auth header name for preconfigured URL
preconfigured_auth_token	Optional[str]	None	Auth token value for preconfigured URL
api_key	Optional[str]	None	API key for health checks (defaults to ENVIRONMENT_API_KEY env var)
health_check_timeout	float	30.0	Max time to wait for health check in seconds
auto_find_port	bool	True	Automatically find available port if requested port is busy
skip_tunnel_verification	bool	True	Skip HTTP verification of tunnel connectivity
force_new_tunnel	bool	False	Force a fresh managed tunnel (Cloudflare managed only)
on_start	Optional[Callable[[InProcessTaskApp], None]]	None	Callback called when task app starts
on_stop	Optional[Callable[[InProcessTaskApp], None]]	None	Callback called when task app stops

Attributes

• url (Optional[str]): The public tunnel URL (available after __aenter__)
• port (int): The actual port the server is running on (may differ from requested if auto_find_port=True)
• host (str): The host the server is bound to
• tunnel_mode (str): The tunnel mode being used
• task_app_worker_token (Optional[str]): SynthTunnel worker token (only set for SynthTunnel)

Raises

• ValueError: If multiple or no input methods provided, or invalid parameters
• FileNotFoundError: If task_app_path doesn’t exist
• RuntimeError: If health check fails or port conflicts can’t be resolved

Usage Examples

Exactly one of app, config, config_factory, or task_app_path must be provided.

1. Direct FastAPI App (app)

from fastapi import FastAPI
from synth_ai.sdk import InProcessTaskApp

app = FastAPI()

@app.get("/health")
def health():
    return {"healthy": True}

async with InProcessTaskApp(app=app, port=8114) as task_app:
    print(f"Running at: {task_app.url}")

2. TaskAppConfig Object (config)

from synth_ai.sdk import InProcessTaskApp
from synth_ai.sdk import TaskAppConfig

config = TaskAppConfig(
    app_id="my_task",
    name="My Task",
    description="A task app",
    base_task_info=task_info,
    describe_taskset=lambda: {"splits": ["train"]},
    provide_task_instances=lambda seeds: [...],
    rollout=lambda req, r: {...},
)

async with InProcessTaskApp(config=config, port=8114) as task_app:
    print(f"Running at: {task_app.url}")

3. Config Factory (config_factory) — Recommended

from synth_ai.sdk import InProcessTaskApp
from heartdisease_task_app import build_config

async with InProcessTaskApp(
    config_factory=build_config,
    port=8114,
) as task_app:
    print(f"Running at: {task_app.url}")

4. File Path (task_app_path)

from synth_ai.sdk import InProcessTaskApp

async with InProcessTaskApp(
    task_app_path="examples/task_apps/banking77/banking77_task_app.py",
    port=8114,
) as task_app:
    print(f"Running at: {task_app.url}")

With Callbacks and Custom Port

from synth_ai.sdk import InProcessTaskApp

def on_start(task_app):
    print(f"✅ Task app started on port {task_app.port}")
    print(f"🌐 Tunnel URL: {task_app.url}")

def on_stop(task_app):
    print(f"🛑 Task app stopped")

async def main():
    async with InProcessTaskApp(
        config_factory=build_config,
        port=9000,
        auto_find_port=True,  # Will find available port if 9000 is busy
        on_start=on_start,
        on_stop=on_stop,
    ) as task_app:
        # Your optimization code here
        pass

asyncio.run(main())

Full GEPA Integration

from synth_ai.sdk import InProcessTaskApp
from synth_ai.sdk import PromptLearningJob
from heartdisease_task_app import build_config

async def main():
    # Start task app in-process
    async with InProcessTaskApp(
        config_factory=build_config,
        port=8114,
    ) as task_app:
        print(f"Task app running at: {task_app.url}")
        
        # Create GEPA job with task app URL
        job = PromptLearningJob(
            backend_url="http://localhost:8000",
            api_key="your-api-key",
        )
        
        # Submit job
        job_id = await job.submit(
            config_path="configs/heartdisease_gepa.toml",
            task_app_url=task_app.url,
        )
        
        # Poll until complete
        results = await job.poll_until_complete(job_id)
        print(f"Best score: {results.best_score}")

asyncio.run(main())

If you’re using SynthTunnel, ensure the job is configured with task_app_worker_token (the SDK can wire this automatically when using run_in_process_job).

Port Conflict Handling

The utility automatically handles port conflicts:

• auto_find_port=True (default): If requested port is busy, automatically finds next available port
• auto_find_port=False: Attempts to kill process on port, then raises error if still busy

# Will automatically find available port if 8114 is busy
async with InProcessTaskApp(
    config_factory=build_config,
    port=8114,
    auto_find_port=True,  # Default
) as task_app:
    print(f"Running on port {task_app.port}")  # May differ from 8114

Input Validation

The utility validates all inputs with clear error messages:

• Port: Must be in range [1024, 65535]
• Host: Must be 127.0.0.1, localhost, or 0.0.0.0 (security requirement)
• Tunnel Mode: "synthtunnel", "quick", "named", "local", or "preconfigured"
• Task App Path: Must exist and be a .py file
• Input Methods: Exactly one of app, config, config_factory, or task_app_path must be provided

Logging

The utility uses Python’s logging module:

import logging

# Configure logging level
logging.basicConfig(level=logging.INFO)

# Now you'll see INFO logs for lifecycle events
async with InProcessTaskApp(...) as task_app:
    pass

Log levels:

• INFO: Major lifecycle events (start, stop, tunnel URL)
• DEBUG: Detailed operations (port checks, health checks)
• WARNING: Port conflicts, callback exceptions

Signal Handling

The utility automatically handles SIGINT/SIGTERM signals for graceful shutdown:

• All active instances are cleaned up on signal
• Prevents orphaned processes
• Works seamlessly with context manager cleanup

Requirements

• Python >= 3.11
• SynthTunnel (default) requires SYNTH_API_KEY
• cloudflared binary only if using Cloudflare tunnels (tunnel_backend="cloudflare_quick")
• Task app must expose /health endpoint
• Task app must accept X-API-Key header for authentication

Troubleshooting

Port Already in Use

If you see “address already in use” errors:

# Option 1: Use auto_find_port (recommended)
async with InProcessTaskApp(..., auto_find_port=True) as task_app:
    pass

# Option 2: Manually kill process on port
# On macOS/Linux:
# lsof -ti:8114 | xargs kill -9

Health Check Timeout

If health check times out:

1. Verify task app has /health endpoint
2. Verify task app accepts X-API-Key header
3. Increase timeout: health_check_timeout=60.0

Tunnel Not Opening

If tunnel fails to open:

1. If using SynthTunnel, confirm SYNTH_API_KEY is set
2. If using Cloudflare, verify cloudflared is installed: which cloudflared
3. Check network connectivity
4. Review logs for detailed error messages

Related Documentation

• Task Apps for Prompt Learning
• Prompt Optimization Cookbook
• SDK Overview