Tork Governance Microservice

Language Agnostic

Call from any language via HTTP

Docker Ready

Containerized deployment

Scalable

Horizontal scaling with load balancing

Observable

Health checks and metrics

# tork_service.py
from typing import Any, Optional
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
import uvicorn

from tork.core.engine import GovernanceEngine
from tork.core.models import EvaluationRequest, PolicyDecision
from tork.pii import PIIRedactor


# Request/Response Models
class EvaluateRequest(BaseModel):
    payload: dict[str, Any]
    agent_id: str = Field(default="external-agent")
    action: str = Field(default="request")


class EvaluateResponse(BaseModel):
    decision: str
    reason: str
    violations: list[str]
    modified_payload: Optional[dict[str, Any]] = None
    pii_detected: list[dict[str, Any]] = []


class RedactRequest(BaseModel):
    text: str
    pii_types: Optional[list[str]] = None


class RedactResponse(BaseModel):
    original: str
    redacted: str
    matches: list[dict[str, Any]]


class HealthResponse(BaseModel):
    status: str
    version: str
    engine_ready: bool


# Initialize services
pii_redactor = PIIRedactor()
engine = GovernanceEngine(
    pii_redactor=pii_redactor,
    enable_auto_redaction=True,
    api_key=None,  # Use local policies or set API key
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize engine
    engine.start()
    yield
    # Shutdown: cleanup
    engine.stop()


app = FastAPI(
    title="Tork Governance Service",
    description="Standalone microservice for AI agent governance",
    version="1.0.0",
    lifespan=lifespan,
)

# Enable CORS for cross-origin requests
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configure for your domain
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check endpoint for load balancers."""
    return HealthResponse(
        status="healthy",
        version="1.0.0",
        engine_ready=engine.is_running(),
    )


@app.get("/ready")
async def readiness_check():
    """Readiness probe for Kubernetes."""
    if not engine.is_running():
        raise HTTPException(status_code=503, detail="Engine not ready")
    return {"ready": True}


@app.post("/evaluate", response_model=EvaluateResponse)
async def evaluate(request: EvaluateRequest):
    """Evaluate payload against governance policies."""
    eval_request = EvaluationRequest(
        payload=request.payload,
        agent_id=request.agent_id,
        action=request.action,
    )

    result = engine.evaluate(eval_request)

    return EvaluateResponse(
        decision=result.decision.value,
        reason=result.reason,
        violations=result.violations,
        modified_payload=result.modified_payload,
        pii_detected=[
            {"type": m.pii_type.value, "start": m.start, "end": m.end}
            for m in result.pii_matches
        ],
    )


@app.post("/redact", response_model=RedactResponse)
async def redact(request: RedactRequest):
    """Redact PII from text."""
    redacted, matches = pii_redactor.redact(request.text)

    return RedactResponse(
        original=request.text,
        redacted=redacted,
        matches=[
            {
                "type": m.pii_type.value,
                "value": m.matched_text,
                "start": m.start,
                "end": m.end,
            }
            for m in matches
        ],
    )


@app.post("/batch/evaluate")
async def batch_evaluate(requests: list[EvaluateRequest]):
    """Evaluate multiple payloads in a single request."""
    results = []
    for req in requests:
        eval_request = EvaluationRequest(
            payload=req.payload,
            agent_id=req.agent_id,
            action=req.action,
        )
        result = engine.evaluate(eval_request)
        results.append({
            "decision": result.decision.value,
            "violations": result.violations,
            "modified_payload": result.modified_payload,
        })
    return {"results": results}


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

# requirements.txt
tork-governance>=0.7.0
fastapi>=0.109.0
uvicorn[standard]>=0.27.0
pydantic>=2.0.0

These clients are yours, not a Tork package

TorkServiceClient below is a thin HTTP wrapper you write yourself against the container you are self-hosting — copy it and adapt it. It is deliberately named to avoid confusion with two real packages: tork-governance (the on-device family, entry class Tork, decides on-device) and @torknetwork/sdk (npm, v2.0.0), whose entry class TorkClient talks to Tork's hosted API rather than to your own container. Because you host and operate this service, the decisions it makes are yours — they are not recorded by Tork as attested_by=tork receipts unless you separately report them.

# docker-compose.yml - Sidecar pattern
version: '3.8'

services:
  # Your main application
  my-app:
    build: ./my-app
    ports:
      - "3000:3000"
    environment:
      - TORK_URL=http://tork-sidecar:8080
    depends_on:
      tork-sidecar:
        condition: service_healthy

  # Tork governance sidecar
  tork-sidecar:
    image: torknetwork/tork-governance:latest
    expose:
      - "8080"
    environment:
      - TORK_API_KEY=${TORK_API_KEY}
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 10s
      timeout: 5s
      retries: 3
    deploy:
      resources:
        limits:
          cpus: '0.5'
          memory: 256M

# Kubernetes sidecar pattern
---
apiVersion: v1
kind: Pod
metadata:
  name: my-app-with-tork
spec:
  containers:
    - name: my-app
      image: my-app:latest
      ports:
        - containerPort: 3000
      env:
        - name: TORK_URL
          value: "http://localhost:8080"

    - name: tork-sidecar
      image: torknetwork/tork-governance:latest
      ports:
        - containerPort: 8080
      resources:
        limits:
          cpu: "500m"
          memory: "256Mi"
      livenessProbe:
        httpGet:
          path: /health
          port: 8080
        initialDelaySeconds: 5
        periodSeconds: 10

# nginx.conf - Load balancing configuration
upstream tork_backend {
    least_conn;
    server tork-1:8080 weight=1;
    server tork-2:8080 weight=1;
    server tork-3:8080 weight=1;

    # Health checks
    keepalive 32;
}

server {
    listen 80;
    server_name tork.internal;

    location / {
        proxy_pass http://tork_backend;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # Timeouts
        proxy_connect_timeout 5s;
        proxy_send_timeout 10s;
        proxy_read_timeout 10s;

        # Retry on failure
        proxy_next_upstream error timeout http_502 http_503 http_504;
        proxy_next_upstream_tries 3;
    }

    location /health {
        proxy_pass http://tork_backend/health;
        proxy_connect_timeout 2s;
        proxy_read_timeout 2s;
    }
}

# Environment variables for configuration

# Tork API key (optional - for cloud policy sync)
TORK_API_KEY=tork_live_xxxxxxxxxxxxx

# Service configuration
TORK_HOST=0.0.0.0
TORK_PORT=8080
TORK_WORKERS=4

# Logging
LOG_LEVEL=info
LOG_FORMAT=json

# Performance tuning
TORK_MAX_PAYLOAD_SIZE=1048576  # 1MB
TORK_REQUEST_TIMEOUT=30
TORK_ENABLE_CACHING=true
TORK_CACHE_TTL=300

# PII Redaction settings
TORK_AUTO_REDACT=true
TORK_PII_TYPES=email,phone,ssn,credit_card

# Policy configuration
TORK_POLICY_DIR=/app/policies
TORK_POLICY_REFRESH_INTERVAL=60

# Enhanced service with Prometheus metrics
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

# Metrics
REQUESTS_TOTAL = Counter(
    'tork_requests_total',
    'Total governance requests',
    ['endpoint', 'decision']
)

REQUEST_LATENCY = Histogram(
    'tork_request_latency_seconds',
    'Request latency in seconds',
    ['endpoint']
)

PII_DETECTIONS = Counter(
    'tork_pii_detections_total',
    'Total PII detections by type',
    ['pii_type']
)

@app.get("/metrics")
async def metrics():
    """Prometheus metrics endpoint."""
    return Response(
        content=generate_latest(),
        media_type="text/plain"
    )

# Add to evaluate endpoint
@app.post("/evaluate", response_model=EvaluateResponse)
async def evaluate(request: EvaluateRequest):
    with REQUEST_LATENCY.labels(endpoint="evaluate").time():
        result = engine.evaluate(...)

    REQUESTS_TOTAL.labels(
        endpoint="evaluate",
        decision=result.decision.value
    ).inc()

    for match in result.pii_matches:
        PII_DETECTIONS.labels(pii_type=match.pii_type.value).inc()

    return result