Python-TypeScript Integration Patterns

┌─────────────────────────────────────────────────────────────────────┐
│                         API Gateway (Hono)                          │
│                        TypeScript / Node.js                         │
└─────────────────────────┬───────────────────────────────────────────┘
                          │
        ┌─────────────────┼─────────────────┐
        │                 │                 │
        ▼                 ▼                 ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ TypeScript    │ │  Python       │ │ TypeScript    │
│ Services      │ │  Services     │ │ Services      │
│ (Veritas,     │ │  (Psyche)     │ │ (Lilith,      │
│  etc.)        │ │               │ │  Hathor)      │
└───────┬───────┘ └───────┬───────┘ └───────┬───────┘
        │                 │                 │
        └─────────────────┼─────────────────┘
                          │
                          ▼
              ┌───────────────────────┐
              │   Shared Services     │
              │  (Redis, PostgreSQL,  │
              │   Qdrant, S3)         │
              └───────────────────────┘

libs/proto/src/
├── common/
│   └── types.proto             # Shared types (UUID, Pagination, Error)
├── health/
│   └── health.proto            # Standard health checking
├── auth/
│   └── auth.proto              # Authentication service
├── isis/
│   └── isis.proto              # Isis generation service
├── sophia/
│   └── sophia.proto            # Sophia knowledge service
├── hathor/
│   └── hathor.proto            # Hathor worldbuilding service
├── bridge/
│   ├── blender.proto           # Blender engine bridge
│   ├── godot.proto             # Godot engine bridge
│   └── unreal.proto            # Unreal engine bridge
├── rendering/
│   └── rendering.proto         # Rendering service
├── ai/
│   └── ai.proto                # AI service definitions
└── [domain]/
    └── [service].proto

// libs/proto/src/psyche/avatar.proto
syntax = "proto3";

package oshun.psyche.avatar;

option go_package = "github.com/oshun/proto/psyche/avatar";

import "google/protobuf/timestamp.proto";
import "common/types.proto";

// Service definition
service AvatarService {
  // Generate a new avatar frame
  rpc GenerateFrame(GenerateFrameRequest) returns (GenerateFrameResponse);

  // Stream avatar frames
  rpc StreamFrames(StreamFramesRequest) returns (stream AvatarFrame);

  // Health check
  rpc Check(oshun.common.HealthCheckRequest) returns (oshun.common.HealthCheckResponse);
}

// Request message
message GenerateFrameRequest {
  string session_id = 1;
  AudioData audio = 2;
  ExpressionState expression = 3;
  RenderSettings settings = 4;
}

// Response message
message GenerateFrameResponse {
  bool success = 1;
  AvatarFrame frame = 2;
  oshun.common.Error error = 3;
}

// Data types
message AvatarFrame {
  bytes image_data = 1;
  string format = 2;  // "png", "jpeg", "webp"
  int32 width = 3;
  int32 height = 4;
  google.protobuf.Timestamp timestamp = 5;
  FrameMetadata metadata = 6;
}

// libs/psyche/client/src/avatar-client.ts
import { credentials, Metadata } from '@grpc/grpc-js';
import { AvatarServiceClient } from '@oshun/proto/psyche/avatar';
import type {
  GenerateFrameRequest,
  GenerateFrameResponse,
} from '@oshun/proto/psyche/avatar';

export class AvatarClient {
  private client: AvatarServiceClient;

  constructor(address: string = 'localhost:50051') {
    this.client = new AvatarServiceClient(
      address,
      credentials.createInsecure()
    );
  }

  async generateFrame(
    request: GenerateFrameRequest,
    options?: { timeout?: number }
  ): Promise<GenerateFrameResponse> {
    const metadata = new Metadata();
    const deadline = new Date();
    deadline.setSeconds(deadline.getSeconds() + (options?.timeout ?? 30));

    return new Promise((resolve, reject) => {
      this.client.generateFrame(
        request,
        metadata,
        { deadline },
        (error, response) => {
          if (error) {
            reject(error);
          } else {
            resolve(response!);
          }
        }
      );
    });
  }

  streamFrames(request: StreamFramesRequest): AsyncIterable<AvatarFrame> {
    const stream = this.client.streamFrames(request);

    return {
      async *[Symbol.asyncIterator]() {
        for await (const frame of stream) {
          yield frame;
        }
      },
    };
  }
}

# services/psyche/avatar-engine/src/avatar_engine/grpc_server.py
import asyncio
from concurrent import futures
from typing import AsyncIterator

import grpc
from grpc_reflection.v1alpha import reflection

from oshun_proto.psyche import avatar_pb2, avatar_pb2_grpc
from oshun_proto.common import types_pb2


class AvatarServicer(avatar_pb2_grpc.AvatarServiceServicer):
    """gRPC service implementation for Avatar generation."""

    def __init__(self, avatar_engine: AvatarEngine):
        self.engine = avatar_engine

    async def GenerateFrame(
        self,
        request: avatar_pb2.GenerateFrameRequest,
        context: grpc.aio.ServicerContext,
    ) -> avatar_pb2.GenerateFrameResponse:
        """Generate a single avatar frame."""
        try:
            frame = await self.engine.generate_frame(
                session_id=request.session_id,
                audio_data=request.audio.data,
                expression=request.expression,
                settings=request.settings,
            )

            return avatar_pb2.GenerateFrameResponse(
                success=True,
                frame=avatar_pb2.AvatarFrame(
                    image_data=frame.data,
                    format=frame.format,
                    width=frame.width,
                    height=frame.height,
                ),
            )
        except Exception as e:
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(str(e))
            return avatar_pb2.GenerateFrameResponse(
                success=False,
                error=types_pb2.Error(
                    code="GENERATION_FAILED",
                    message=str(e),
                ),
            )

    async def StreamFrames(
        self,
        request: avatar_pb2.StreamFramesRequest,
        context: grpc.aio.ServicerContext,
    ) -> AsyncIterator[avatar_pb2.AvatarFrame]:
        """Stream avatar frames for real-time rendering."""
        async for frame in self.engine.stream_frames(
            session_id=request.session_id,
            audio_stream=request.audio_stream,
        ):
            if context.cancelled():
                break
            yield avatar_pb2.AvatarFrame(
                image_data=frame.data,
                format=frame.format,
                width=frame.width,
                height=frame.height,
            )


async def serve(port: int = 50051):
    """Start the gRPC server."""
    server = grpc.aio.server(futures.ThreadPoolExecutor(max_workers=10))

    # Add servicer
    avatar_engine = AvatarEngine()
    avatar_pb2_grpc.add_AvatarServiceServicer_to_server(
        AvatarServicer(avatar_engine), server
    )

    # Enable reflection for debugging
    SERVICE_NAMES = (
        avatar_pb2.DESCRIPTOR.services_by_name['AvatarService'].full_name,
        reflection.SERVICE_NAME,
    )
    reflection.enable_server_reflection(SERVICE_NAMES, server)

    server.add_insecure_port(f'[::]:{port}')
    await server.start()
    await server.wait_for_termination()


if __name__ == '__main__':
    asyncio.run(serve())

# Generate TypeScript types and client
nx build @oshun/proto

# Generate Python types (in each Python service)
cd services/psyche/avatar-engine
poetry run python -m grpc_tools.protoc \
  -I../../../libs/proto/src \
  --python_out=./src \
  --pyi_out=./src \
  --grpc_python_out=./src \
  ../../../libs/proto/src/psyche/avatar.proto

// Old: package oshun.psyche.avatar.v1;
// New: package oshun.psyche.avatar.v2;

// libs/psyche/client/src/http-client.ts
import { httpClient } from '@oshun/http-client';
import type { AvatarSession, VoiceConfig } from '@psyche/contracts';

export class PsycheHttpClient {
  private baseUrl: string;

  constructor(baseUrl: string = process.env.PSYCHE_API_URL!) {
    this.baseUrl = baseUrl;
  }

  async createSession(config: SessionConfig): Promise<AvatarSession> {
    const response = await httpClient.post<AvatarSession>(
      `${this.baseUrl}/sessions`,
      config
    );
    return response.data;
  }

  async getVoices(): Promise<VoiceConfig[]> {
    const response = await httpClient.get<VoiceConfig[]>(
      `${this.baseUrl}/voices`
    );
    return response.data;
  }
}

# libs/psyche/common/src/common/clients/oshun_client.py
import httpx
from pydantic import BaseModel
from typing import TypeVar, Generic

T = TypeVar("T", bound=BaseModel)


class OshunServiceClient:
    """HTTP client for Oshun TypeScript services."""

    def __init__(
        self,
        base_url: str,
        api_key: str | None = None,
        timeout: float = 30.0,
    ):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.headers = {"Content-Type": "application/json"}
        if api_key:
            self.headers["Authorization"] = f"Bearer {api_key}"

    async def get(self, path: str, response_model: type[T]) -> T:
        """GET request with response parsing."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(
                f"{self.base_url}{path}",
                headers=self.headers,
            )
            response.raise_for_status()
            return response_model.model_validate(response.json())

    async def post(
        self,
        path: str,
        data: BaseModel,
        response_model: type[T],
    ) -> T:
        """POST request with request/response parsing."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                f"{self.base_url}{path}",
                headers=self.headers,
                json=data.model_dump(mode="json"),
            )
            response.raise_for_status()
            return response_model.model_validate(response.json())


# Usage example
class AuthClient(OshunServiceClient):
    """Client for @oshun/auth service."""

    def __init__(self):
        super().__init__(
            base_url=os.environ["AUTH_SERVICE_URL"],
            api_key=os.environ.get("AUTH_SERVICE_API_KEY"),
        )

    async def validate_token(self, token: str) -> TokenValidation:
        return await self.post(
            "/validate",
            TokenValidateRequest(token=token),
            TokenValidation,
        )

<DOMAIN>_<SERVICE>_<SETTING>=value

# Service URLs
PSYCHE_AVATAR_URL=http://localhost:8001
PSYCHE_VOICE_URL=http://localhost:8002
VERITAS_API_URL=http://localhost:3001

# gRPC Ports
PSYCHE_AVATAR_GRPC_PORT=50051
PSYCHE_VOICE_GRPC_PORT=50052

# Database
PSYCHE_DATABASE_URL=postgresql://user:pass@localhost:5432/psyche
VERITAS_DATABASE_URL=postgresql://user:pass@localhost:5432/veritas

# Redis
REDIS_URL=redis://localhost:6379

# API Keys (shared)
ANTHROPIC_API_KEY=sk-ant-xxx
OPENAI_API_KEY=sk-xxx
ELEVENLABS_API_KEY=xxx

// apps/veritas/api/src/config.ts
import { z } from 'zod';

const configSchema = z.object({
  // Server
  PORT: z.coerce.number().default(3001),
  HOST: z.string().default('0.0.0.0'),
  NODE_ENV: z
    .enum(['development', 'production', 'test'])
    .default('development'),

  // Database
  DATABASE_URL: z.string().url(),

  // Redis
  REDIS_URL: z.string().url(),

  // Psyche integration
  PSYCHE_AVATAR_URL: z.string().url(),
  PSYCHE_VOICE_URL: z.string().url(),
  PSYCHE_AVATAR_GRPC_PORT: z.coerce.number().default(50051),

  // API Keys
  ANTHROPIC_API_KEY: z.string().optional(),
  OPENAI_API_KEY: z.string().optional(),
});

export const config = configSchema.parse(process.env);

# services/psyche/avatar-engine/src/avatar_engine/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    """Avatar Engine configuration."""

    model_config = SettingsConfigDict(
        env_prefix="PSYCHE_AVATAR_",
        env_file=".env",
        case_sensitive=False,
    )

    # Server
    host: str = "0.0.0.0"
    port: int = 8001
    grpc_port: int = 50051
    workers: int = 4

    # Database
    database_url: str

    # Redis
    redis_url: str

    # Model paths
    model_path: str = "/models/avatar"
    cache_dir: str = "/cache"

    # GPU
    cuda_device: int = 0
    enable_cuda: bool = True

    # External services
    voice_service_url: str = "http://localhost:8002"
    auth_service_url: str = "http://localhost:3000"

    # API Keys
    anthropic_api_key: str | None = None


settings = Settings()

# docker/docker-compose.dev.yml uses:
# .env                    - Root environment file
# .env.local              - Local overrides (gitignored)
# services/psyche/.env    - Psyche-specific defaults

// libs/shared/errors/src/service-error.ts
export class ServiceError extends Error {
  constructor(
    public code: string,
    message: string,
    public statusCode: number = 500,
    public details?: Record<string, unknown>
  ) {
    super(message);
    this.name = 'ServiceError';
  }

  toJSON() {
    return {
      code: this.code,
      message: this.message,
      details: this.details,
    };
  }
}

// Usage
throw new ServiceError(
  'PSYCHE_3001',
  'Avatar generation failed: GPU memory exceeded',
  503,
  { requiredMemory: '24GB', available: '12GB' }
);

# libs/psyche/common/src/common/errors.py
from dataclasses import dataclass
from typing import Any


@dataclass
class ServiceError(Exception):
    """Standard service error."""

    code: str
    message: str
    status_code: int = 500
    details: dict[str, Any] | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "code": self.code,
            "message": self.message,
            "details": self.details,
        }


class PsycheError(ServiceError):
    """Psyche domain errors."""

    def __init__(
        self,
        code: str,
        message: str,
        status_code: int = 500,
        details: dict[str, Any] | None = None,
    ):
        super().__init__(
            code=f"PSYCHE_{code}",
            message=message,
            status_code=status_code,
            details=details,
        )


# Usage
raise PsycheError(
    code="3001",
    message="GPU memory exceeded",
    status_code=503,
    details={"required": "24GB", "available": "12GB"},
)

┌──────────┐    ┌──────────────┐    ┌──────────────┐
│  Client  │───▶│  API Gateway │───▶│   @oshun/    │
│          │    │  (TypeScript)│    │    auth      │
└──────────┘    └──────┬───────┘    └──────────────┘
                       │
                       │ JWT Token
                       ▼
              ┌────────────────┐
              │ Python Service │
              │ (Psyche)       │
              │                │
              │ Validates JWT  │
              │ via shared     │
              │ secret/JWKS    │
              └────────────────┘

# services/psyche/api-gateway/src/api_gateway/auth.py
import jwt
from fastapi import HTTPException, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from functools import lru_cache

security = HTTPBearer()


@lru_cache()
def get_jwt_secret() -> str:
    """Get JWT secret from environment or secrets manager."""
    return os.environ["JWT_SECRET"]


async def verify_token(
    credentials: HTTPAuthorizationCredentials = Security(security),
) -> dict:
    """Verify JWT token and return payload."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            get_jwt_secret(),
            algorithms=["HS256"],
            options={"verify_exp": True},
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=401,
            detail={"code": "AUTH_1002", "message": "Token expired"},
        )
    except jwt.InvalidTokenError as e:
        raise HTTPException(
            status_code=401,
            detail={"code": "AUTH_1001", "message": f"Invalid token: {e}"},
        )


# Usage in routes
@router.get("/sessions")
async def list_sessions(
    user: dict = Depends(verify_token),
):
    user_id = user["sub"]
    return await session_service.list_by_user(user_id)

// libs/shared/tracing/src/index.ts
import { trace, SpanStatusCode } from '@opentelemetry/api';

const tracer = trace.getTracer('veritas-api');

export async function withSpan<T>(
  name: string,
  fn: () => Promise<T>
): Promise<T> {
  return tracer.startActiveSpan(name, async (span) => {
    try {
      const result = await fn();
      span.setStatus({ code: SpanStatusCode.OK });
      return result;
    } catch (error) {
      span.setStatus({ code: SpanStatusCode.ERROR, message: String(error) });
      throw error;
    } finally {
      span.end();
    }
  });
}

# services/psyche/avatar-engine/src/avatar_engine/tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

# Initialize
provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint=os.environ.get("OTLP_ENDPOINT", "localhost:4317"))
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("psyche-avatar-engine")


# Usage
async def generate_frame(session_id: str, audio: bytes) -> Frame:
    with tracer.start_as_current_span("generate_frame") as span:
        span.set_attribute("session.id", session_id)
        span.set_attribute("audio.size", len(audio))

        # ... generation logic

        return frame