In-Process Task App

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())

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}")

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}")

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}")

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}")

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())

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())

# 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

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

# 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