> ## Documentation Index
> Fetch the complete documentation index at: https://docs.usesynth.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# 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.

<Note>
  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.
</Note>

## Quick Start

```python theme={null}
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

<CardGroup cols={2}>
  <Card title="Automatic Server Management" icon="server">
    Starts uvicorn server in background thread
  </Card>

  <Card title="Automatic Tunnel Creation" icon="cloud">
    Opens SynthTunnel by default (Cloudflare optional)
  </Card>

  <Card title="Port Conflict Handling" icon="settings">
    Automatically finds available ports if requested port is busy
  </Card>

  <Card title="Signal Handling" icon="shield">
    Graceful shutdown on SIGINT/SIGTERM
  </Card>

  <Card title="Observability" icon="chart">
    Structured logging and optional callbacks
  </Card>

  <Card title="Multiple Input Methods" icon="code">
    Supports app, config, config\_factory, or file path
  </Card>
</CardGroup>

## 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

<Note>Exactly **one** of `app`, `config`, `config_factory`, or `task_app_path` must be provided.</Note>

### 1. Direct FastAPI App (`app`)

```python theme={null}
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`)

```python theme={null}
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

```python theme={null}
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`)

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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())
```

<Note>
  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`).
</Note>

## 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

```python theme={null}
# 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:

```python theme={null}
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:

```python theme={null}
# 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](/sdk/localapi/in-process)
* [Prompt Optimization Cookbook](/sdk/localapi/in-process)
* [SDK Overview](/sdk/overview)