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
#
# Every name imported from tork_governance below exists in the published
# tork-governance 0.26.1 package (PyPI). Governance runs on-device inside this
# process: no request leaves the container, no API key is needed.
import os
from typing import Any, Optional

from fastapi import FastAPI
from pydantic import BaseModel, Field
import uvicorn

from tork_governance import GovernanceAction, GovernanceResult, Tork, __version__ as TORK_VERSION


# Request/Response Models
class GovernRequest(BaseModel):
    content: str
    agent_id: Optional[str] = Field(default=None)
    region: Optional[list[str]] = Field(default=None)    # e.g. ["ZA", "EU"]
    industry: Optional[str] = Field(default=None)        # e.g. "healthcare"


class GovernResponse(BaseModel):
    action: str                     # "allow" | "redact" | "deny"
    output: str                     # redacted text when action == "redact"
    pii: dict[str, Any]             # {"has_pii": bool, "types": [...], "count": int}
    receipt: dict[str, Any]         # local receipt, see to_receipt_dict()


class RedactRequest(BaseModel):
    text: str


class RedactResponse(BaseModel):
    original: str
    redacted: str
    types: list[str]


class HealthResponse(BaseModel):
    status: str
    sdk_version: str


# One client per action mode. TORK_DEFAULT_ACTION decides what a PII hit does
# on /govern: "redact" (default) rewrites the text, "deny" leaves it untouched
# and reports action="deny" so the caller can block.
POLICY_VERSION = os.environ.get("TORK_POLICY_VERSION", "1.0.0")
DEFAULT_ACTION = GovernanceAction(os.environ.get("TORK_DEFAULT_ACTION", "redact"))

governor = Tork(policy_version=POLICY_VERSION, default_action=DEFAULT_ACTION)
redactor = Tork(policy_version=POLICY_VERSION, default_action=GovernanceAction.REDACT)


def to_receipt_dict(result: GovernanceResult) -> dict[str, Any]:
    r = result.receipt
    return {
        "receipt_id": r.receipt_id,
        "timestamp": r.timestamp,
        "action": r.action.value,
        "input_hash": r.input_hash,
        "output_hash": r.output_hash,
        "policy_version": r.policy_version,
        "processing_time_ns": r.processing_time_ns,
        "pii_types": [t.value for t in r.pii_types],
    }


def to_govern_response(result: GovernanceResult) -> GovernResponse:
    return GovernResponse(
        action=result.action.value,
        output=result.output,
        pii={
            "has_pii": result.pii.has_pii,
            "types": [t.value for t in result.pii.types],
            "count": result.pii.count,
        },
        receipt=to_receipt_dict(result),
    )


app = FastAPI(
    title="Tork Governance Service",
    description="Self-hosted HTTP wrapper around the on-device tork-governance SDK",
    version="1.0.0",
)


@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check endpoint for load balancers."""
    return HealthResponse(status="healthy", sdk_version=TORK_VERSION)


@app.get("/ready")
async def readiness_check():
    """Readiness probe for Kubernetes. The SDK has no warm-up step."""
    return {"ready": True}


@app.post("/govern", response_model=GovernResponse)
async def govern(request: GovernRequest):
    """Govern one string: detect PII, apply the default action, mint a local receipt."""
    result = governor.govern(
        request.content,
        region=request.region,
        industry=request.industry,
        agent_id=request.agent_id,
    )
    return to_govern_response(result)


@app.post("/redact", response_model=RedactResponse)
async def redact(request: RedactRequest):
    """Always redact, whatever TORK_DEFAULT_ACTION is."""
    result = redactor.govern(request.text)
    return RedactResponse(
        original=request.text,
        redacted=result.output,
        types=[t.value for t in result.pii.types],
    )


@app.post("/batch/govern")
async def batch_govern(requests: list[GovernRequest]):
    """Govern several strings in one call. Each item gets its own receipt."""
    return {
        "results": [
            to_govern_response(
                governor.govern(
                    req.content,
                    region=req.region,
                    industry=req.industry,
                    agent_id=req.agent_id,
                )
            )
            for req in requests
        ]
    }


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

# requirements.txt
tork-governance>=0.26.1
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 — the image you build from the Dockerfile above
  # (there is no published torknetwork/tork-governance image on Docker Hub)
  tork-sidecar:
    build: ./tork-service
    expose:
      - "8080"
    environment:
      - TORK_DEFAULT_ACTION=redact
    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: your-registry/tork-governance:latest   # built from the Dockerfile above
      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 read by tork_service.py

# Recorded on every receipt
TORK_POLICY_VERSION=1.0.0

# What a PII hit does on /govern: redact (default) or deny
TORK_DEFAULT_ACTION=redact

# No TORK_API_KEY: the SDK governs on-device and never calls tork.network.
# Host, port and worker count are uvicorn arguments (see the Dockerfile CMD).

# 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', 'action']
)

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

# Instrumented /govern (replaces the plain handler above)
@app.post("/govern", response_model=GovernResponse)
async def govern(request: GovernRequest):
    with REQUEST_LATENCY.labels(endpoint="govern").time():
        result = governor.govern(
            request.content,
            region=request.region,
            industry=request.industry,
            agent_id=request.agent_id,
        )

    REQUESTS_TOTAL.labels(
        endpoint="govern",
        action=result.action.value
    ).inc()

    for pii_type in result.pii.types:
        PII_DETECTIONS.labels(pii_type=pii_type.value).inc()

    return to_govern_response(result)