TealRegistry

pip install tealtiger

import asyncio
from tealtiger import TealEngine, TealRegistry

engine = TealEngine(
    policy={
        "registry": {
            "models": ["gpt-4", "gpt-4o", "claude-3-sonnet"],
            "tools": ["web_search", "file_read", "database_query"],
        }
    },
    modules=[TealRegistry()],
)

async def main():
    # Allowed model + tool
    decision = await engine.evaluate(
        request={"model": "gpt-4", "tool": "web_search"},
        ctx={"correlation_id": "req-001"},
    )
    print(decision.action)  # "ALLOW"

    # Unknown model — denied
    denied = await engine.evaluate(
        request={"model": "unknown-model-v1"},
        ctx={"correlation_id": "req-002"},
    )
    print(denied.action)        # "DENY"
    print(denied.reason_codes)  # ["MODEL_NOT_ALLOWLISTED"]

asyncio.run(main())

class TealRegistry:
    def __init__(
        self,
        strict: bool = True,
        verify_provenance: bool = False,
    ) -> None: ...

engine = TealEngine(
    policy={
        "registry": {
            "models": ["gpt-4", "gpt-4o", "gpt-4o-mini", "claude-3-sonnet"],
            "tools": ["web_search", "file_read", "database_query"],
            "strict": True,
            "verify_provenance": True,
        }
    },
    modules=[TealRegistry()],
)

engine = TealEngine(
    policy={
        "registry": {
            "models": [
                {"id": "gpt-4", "version": "0613", "hash": "sha256:abc..."},
                {"id": "claude-3-sonnet", "version": "20240229"},
            ],
            "tools": [
                {"id": "web_search", "version": "2.1.0"},
            ],
        }
    },
    modules=[TealRegistry()],
)

from dataclasses import dataclass

@dataclass
class RegistryRef:
    catalog: str   # "models" | "tools" | "detectors" | "policies"
    entry_id: str  # Unique entry identifier
    version: str   # Pinned version
    hash: str      # Content hash for integrity verification

from tealtiger import TealRegistry

registry = TealRegistry(verify_provenance=True)

drift_result = await registry.check_definition_drift(
    agent_id="agent-001",
    current_tools=[
        {"name": "web_search", "parameters": {"query": "string"}},
        {"name": "file_read", "parameters": {"path": "string"}},
        {"name": "shell_execute", "parameters": {"command": "string"}},  # new!
    ],
)

if drift_result.drifted:
    for change in drift_result.changes:
        print(f"{change.type}: {change.tool} (risk={change.risk})")

decision = await engine.evaluate(
    request={"model": "gpt-4", "tool": "execute_shell_command"},
    ctx={"correlation_id": "req-003"},
)

print(decision.action)        # "DENY"
print(decision.reason_codes)  # ["TOOL_NOT_ALLOWED"]

# Hot-reload the registry via policy swap
await engine.load_policy({
    "version": "2.0.0",
    "rules": [],
    "modules": {
        "registry": {
            "models": ["gpt-4", "gpt-4o", "claude-3.5-sonnet", "gemini-1.5-pro"],
            "tools": ["web_search", "file_read", "code_interpreter"],
        }
    },
})