pydantic-deep-worker

Worker service for distributed pydantic-ai agent execution. Registers with a gateway and processes subagent requests.

Overview

The worker provides:

• Automatic registration with the gateway on startup
• Subagent request handling and execution
• Health check endpoint for gateway monitoring
• Support for multiple subagent types per worker

Installation

cd pydantic-deep-worker
make install

Quick Start

# Run worker (default: port 8001)
WORKER_ID=worker-1 GATEWAY_URL=http://gateway:8080 uv run python main.py

Environment Variables

Variable	Default	Description
WORKER_HOST	0.0.0.0	Host to bind worker service
WORKER_PORT	8001	Port for worker API
WORKER_ID	default	Unique identifier for this worker
SUBAGENT_TYPES	general-purpose	Comma-separated list of subagent types
HEALTH_URL	Auto-generated	URL for health check endpoint
API_URL	Auto-generated	URL for subagent execution endpoint
GATEWAY_URL	http://localhost:8080	Gateway URL for registration
MODEL	openai:gpt-4.1	Model to use for subagent execution
RUNTIME_NAME	-	Optional runtime name for sandbox execution

API Endpoints

Worker Management

• GET /health - Worker health check (returns worker_id and status)

Subagent Execution

• POST /run - Execute subagent request

Usage as a Library

from pydantic_deep_worker import WorkerService

worker = WorkerService(
    worker_id="worker-1",
    subagent_types=["code-review", "debugging"],
    health_url="http://worker-1:8001/health",
    api_url="http://worker-1:8001",
    gateway_url="http://gateway:8080",
)
app = worker.create_app()

# Run with: uvicorn app:app --host 0.0.0.0 --port 8001

Request/Response Format

SubAgentRequest

{
    "description": "Task description",
    "subagent_type": "code-review",
    "session_id": "session-123",
    "trace_id": "trace-456"  # optional
}

SubAgentResponse

{
    "output": "Result from subagent execution",
    "error": null,  # or error message
    "trace_id": "trace-456"
}

Architecture

Workers follow a simple lifecycle:

1. Startup - Register with gateway via POST /api/v1/worker/register
2. Running - Receive and process subagent requests via POST /run
3. Health Checks - Gateway periodically checks worker health via GET /health
4. Shutdown - Unregister from gateway via DELETE /api/v1/worker/unregister/{worker_id}

Development

make install      # Install dependencies
make test         # Run tests
make testcov      # Run tests with coverage
make format       # Format code
make lint         # Run linter
make typecheck    # Run type checking (pyright + mypy)
make all          # Run all checks

Docker Deployment

FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN make install
CMD ["python", "main.py"]

docker run -e WORKER_ID=worker-1 \
           -e SUBAGENT_TYPES=code-review \
           -e GATEWAY_URL=http://gateway:8080 \
           -p 8001:8001 \
           pydantic-deep-worker

License

MIT