API Services

This guide covers building backend services, REST APIs, batch processing pipelines, and CLI tools with the Zeq OS Python SDK. It also covers integrating with the API Gateway on port 4000 for authentication, rate limiting, and the full HTTP API surface.

Setup

pip install zeq-os

from zeq_os import ZeqProcessor, HulyaSync, OperatorRegistry

processor = ZeqProcessor()
sync = HulyaSync()
registry = OperatorRegistry.default()

print(f"{registry.count} operators across {len(registry.categories)} categories")
print(sync.daemon_tick())

FastAPI Service

A complete FastAPI service that exposes physics computation, operator lookup, and HulyaPulse status:

from fastapi import FastAPI, Query, HTTPException
from zeq_os import ZeqProcessor, HulyaSync, OperatorRegistry

app = FastAPI(
    title="Zeq OS Physics Service",
    version="1.287.5",
    description="Physics computation API powered by 1,576 operators"
)

processor = ZeqProcessor()
sync = HulyaSync()
registry = OperatorRegistry.default()


@app.post("/compute")
def compute(query: str, domain_hints: list[str] = Query(default=None)):
    """Process a physics query through the 7-Step Wizard Protocol."""
    state = processor.process(query, domain_hints=domain_hints)
    return {
        "query": query,
        "domains": state.domains,
        "operators": state.selected_operators,
        "operator_count": len(state.selected_operators),
        "master_sum": state.master_sum,
        "phase_coherence": state.phase_coherence,
        "zeqond": state.zeqond,
        "phase": state.phase,
        "precision_met": state.phase_coherence <= 0.1,
    }


@app.get("/operators")
def list_operators(category: str = None):
    """List operators, optionally filtered by category."""
    if category:
        ops = registry.find_by_category(category)
        if not ops:
            raise HTTPException(status_code=404, detail=f"Category '{category}' not found")
    else:
        ops = registry.all()
    return {
        "count": len(ops),
        "operators": [
            {"id": o.id, "category": o.category, "description": o.description}
            for o in ops
        ],
    }


@app.get("/operators/{operator_id}")
def get_operator(operator_id: str):
    """Get a single operator by ID."""
    op = registry.find_by_id(operator_id)
    if not op:
        raise HTTPException(status_code=404, detail=f"Operator '{operator_id}' not found")
    return {
        "id": op.id,
        "category": op.category,
        "description": op.description,
        "equation": op.equation,
    }


@app.get("/pulse")
def pulse():
    """Current HulyaPulse state."""
    return {
        "zeqond": sync.get_zeqond(),
        "phase": sync.current_phase(),
        "ko42": sync.ko42_automatic(),
        "frequency": 1.287,
        "tick": sync.daemon_tick(),
    }

Run with:

uvicorn my_service:app --host 0.0.0.0 --port 8080

Batch Processing

Process multiple physics queries in a loop. Each computation is independently synchronized to HulyaPulse and includes KO42 automatically:

from zeq_os import ZeqProcessor

processor = ZeqProcessor()

queries = [
    ("Calculate orbital velocity at 400km altitude", ["aerospace"]),
    ("Quantum tunneling probability for electron at 5eV barrier", ["quantum"]),
    ("Gravitational time dilation at r=10km from Earth", ["relativity"]),
    ("Black-Scholes pricing for European call option", ["finance"]),
    ("Drug half-life for 500mg acetaminophen", ["medical"]),
]

results = []
for query, hints in queries:
    state = processor.process(query, domain_hints=hints)
    results.append({
        "query": query,
        "operators": state.selected_operators,
        "master_sum": state.master_sum,
        "coherence": state.phase_coherence,
        "precision_met": state.phase_coherence <= 0.1,
    })
    print(f"[OK] {query[:50]}... -> {state.selected_operators}")

# Summary
passed = sum(1 for r in results if r["precision_met"])
print(f"\n{passed}/{len(results)} computations met ≤0.1% precision target")

Pipeline Pattern

Chain results from one operator computation into the next. This is useful when a later calculation depends on values produced by an earlier one:

from zeq_os import ZeqProcessor

processor = ZeqProcessor()

# Step 1: Calculate orbital velocity
orbit = processor.process(
    "Orbital velocity at 400km altitude above Earth",
    domain_hints=["aerospace"]
)
print(f"Step 1 — Operators: {orbit.selected_operators}")
print(f"         Master Sum: {orbit.master_sum:.6f}")

# Step 2: Use the orbital result to calculate relativistic effects
relativistic = processor.process(
    f"Relativistic time dilation at velocity {orbit.master_sum}c",
    domain_hints=["relativity"]
)
print(f"Step 2 — Operators: {relativistic.selected_operators}")
print(f"         Master Sum: {relativistic.master_sum:.6f}")

# Step 3: Apply quantum corrections
corrected = processor.process(
    f"Quantum correction to gravitational field with dilation factor {relativistic.master_sum}",
    domain_hints=["quantum"]
)
print(f"Step 3 — Operators: {corrected.selected_operators}")
print(f"         Master Sum: {corrected.master_sum:.6f}")
print(f"         Precision Met: {corrected.phase_coherence <= 0.1}")

Each step independently runs the full 7-Step Protocol with KO42 as the mandatory first operator. The pipeline chains semantic context, not raw operator state — each process() call is a fresh computation.

CLI Tool

Build a command-line tool for quick physics calculations using argparse:

#!/usr/bin/env python3
"""zeq-compute: CLI for Zeq OS physics computation."""

import argparse
import json
from zeq_os import ZeqProcessor, HulyaSync, OperatorRegistry

def main():
    parser = argparse.ArgumentParser(description="Zeq OS Physics CLI")
    subparsers = parser.add_subparsers(dest="command")

    # process command
    proc = subparsers.add_parser("process", help="Process a physics query")
    proc.add_argument("query", help="Natural-language physics query")
    proc.add_argument("--domain", "-d", action="append", help="Domain hint (repeatable)")
    proc.add_argument("--json", action="store_true", help="Output raw JSON")

    # operators command
    ops = subparsers.add_parser("operators", help="List operators")
    ops.add_argument("--category", "-c", help="Filter by category")
    ops.add_argument("--search", "-s", help="Search by keyword")

    # pulse command
    subparsers.add_parser("pulse", help="Show current HulyaPulse state")

    args = parser.parse_args()

    if args.command == "process":
        processor = ZeqProcessor()
        state = processor.process(args.query, domain_hints=args.domain)

        if args.json:
            print(json.dumps({
                "domains": state.domains,
                "operators": state.selected_operators,
                "master_sum": state.master_sum,
                "phase_coherence": state.phase_coherence,
                "zeqond": state.zeqond,
            }, indent=2))
        else:
            print(f"Domains:    {', '.join(state.domains)}")
            print(f"Operators:  {', '.join(state.selected_operators)}")
            print(f"Master Sum: {state.master_sum:.6f}")
            print(f"Coherence:  {state.phase_coherence:.4f}%")
            print(f"Zeqond:     {state.zeqond}")
            ok = "PASS" if state.phase_coherence <= 0.1 else "WARN"
            print(f"Precision:  {ok}")

    elif args.command == "operators":
        registry = OperatorRegistry.default()
        if args.search:
            ops = registry.search(args.search)
        elif args.category:
            ops = registry.find_by_category(args.category)
        else:
            ops = registry.all()
        for op in ops:
            print(f"  {op.id:25s} {op.category:20s} {op.description}")
        print(f"\n{len(ops)} operators")

    elif args.command == "pulse":
        sync = HulyaSync()
        print(f"Zeqond:    {sync.get_zeqond()}")
        print(f"Phase:     {sync.current_phase():.4f}")
        print(f"KO42:      {sync.ko42_automatic():.6f}")
        print(f"Frequency: 1.287 Hz")
        print(f"Tick:      {sync.daemon_tick()}")

    else:
        parser.print_help()


if __name__ == "__main__":
    main()

Usage:

# Process a query
python zeq_compute.py process "gravitational time dilation at r=10km" -d relativity

# JSON output for piping
python zeq_compute.py process "orbital velocity at 400km" --json | jq .operators

# List quantum operators
python zeq_compute.py operators -c quantum

# Search for tunneling-related operators
python zeq_compute.py operators -s tunneling

# Check current pulse
python zeq_compute.py pulse

Integrating with the API Gateway

The API Gateway runs on port 4000 and provides the full Zeq OS API over HTTP. Use it when your service needs to communicate with Zeq OS over the network rather than importing the SDK directly.

Authentication

All /api/* endpoints (except /api/v1/status and /api-health) require an API key:

# Set your API key
export ZEQ_API_KEY="your-api-key-here"

# Via X-API-Key header
curl -H "X-API-Key: $ZEQ_API_KEY" http://localhost:4000/api/v1/operators/all

# Via Authorization header
curl -H "Authorization: Bearer $ZEQ_API_KEY" http://localhost:4000/api/v1/operators/all

API keys are configured on the gateway via the ZEQ_API_KEYS environment variable (comma-separated for multiple keys).

Rate Limiting

Default: 100 requests/minute per IP. Configurable via RATE_LIMIT_RPM. Rate limit headers are included in every response:

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1740499260

API Gateway Endpoints

Method	Endpoint	Description
GET	/api-health	Gateway health (no auth)
GET	/api/v1/status	Extended status with operator count (no auth)
GET	/api/v1/operators/all	All 1,576 operators
GET	/api/v1/operators/:id	Single operator by ID
GET	/api/v1/operators/category/:category	Filter by category
GET	/api/v1/operators/search?q=:query	Full-text search
GET	/api/v1/categories	All 64 categories with counts
POST	/api/v1/process	Process a physics query

curl Examples

# Health check (no auth required)
curl http://localhost:4000/api-health
# {"status":"ok","service":"zeq-api-gateway","version":"1.287.5","uptime":86400}

# Status with operator count (no auth required)
curl http://localhost:4000/api/v1/status
# {"status":"ok","operators":1576,"categories":64,"version":"6.3.0","pulse":{"frequency":1.287,"zeqond":0.777}}

# Get all operators
curl -H "X-API-Key: $ZEQ_API_KEY" http://localhost:4000/api/v1/operators/all

# Get a specific operator
curl -H "X-API-Key: $ZEQ_API_KEY" http://localhost:4000/api/v1/operators/KO42

# Search for operators
curl -H "X-API-Key: $ZEQ_API_KEY" "http://localhost:4000/api/v1/operators/search?q=tunneling"

# Filter by category
curl -H "X-API-Key: $ZEQ_API_KEY" http://localhost:4000/api/v1/operators/category/quantum

# List all categories
curl -H "X-API-Key: $ZEQ_API_KEY" http://localhost:4000/api/v1/categories

# Process a physics query
curl -X POST http://localhost:4000/api/v1/process \
  -H "X-API-Key: $ZEQ_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "query": "Calculate gravitational time dilation at r=10km from Earth",
    "domain_hints": ["relativity"],
    "precision": 0.001
  }'

Calling the Gateway from Python

If your backend service needs to call the API Gateway rather than importing the SDK:

import requests

GATEWAY_URL = "http://localhost:4000"
API_KEY = "your-api-key"
HEADERS = {"X-API-Key": API_KEY}


def process_query(query: str, domain_hints: list[str] = None) -> dict:
    """Process a physics query via the API Gateway."""
    response = requests.post(
        f"{GATEWAY_URL}/api/v1/process",
        headers=HEADERS,
        json={"query": query, "domain_hints": domain_hints},
    )
    response.raise_for_status()
    return response.json()["state"]


def get_operator(operator_id: str) -> dict:
    """Look up a single operator by ID."""
    response = requests.get(
        f"{GATEWAY_URL}/api/v1/operators/{operator_id}",
        headers=HEADERS,
    )
    response.raise_for_status()
    return response.json()


# Example usage
result = process_query("orbital velocity at 400km", domain_hints=["aerospace"])
print(f"Operators: {result['selected_operators']}")
print(f"Precision: {'PASS' if result['precision_met'] else 'FAIL'}")

ko42 = get_operator("KO42")
print(f"KO42: {ko42['description']}")

Next Steps

• Industry Examples — Real-world examples across 6 domains
• Building Applications Overview — Architecture and requirements
• API Gateway Reference — Full endpoint documentation
• Python SDK — Complete Python API reference
• HITE Encryption — Encrypt API traffic with HITE