Skip to content

Python SDK

PyHall is the Python reference implementation of WCP. It implements WCP-Full — the complete compliance tier including blast radius scoring, privilege envelopes, policy gate evaluation, and evidence receipts.

Version: 0.3.0 | PyPI: pyhall-wcp | WCP spec: 0.1

Install

Terminal window
pip install pyhall-wcp

Quick Start

import uuid
from pyhall import make_decision, RouteInput, Registry, load_rules
# Load routing rules and enrolled workers
rules = load_rules("rules.json")
registry = Registry(registry_dir="enrolled/")
# Build a capability request
inp = RouteInput(
capability_id="cap.doc.summarize",
env="dev",
data_label="INTERNAL",
tenant_risk="low",
qos_class="P2",
tenant_id="acme-corp",
correlation_id=str(uuid.uuid4()),
)
# Ask the Hall
decision = make_decision(
inp=inp,
rules=rules,
registry_controls_present=registry.controls_present(),
registry_worker_available=registry.worker_available,
)
if decision.denied:
print(f"Denied: {decision.deny_reason_if_denied}")
else:
print(f"Dispatch to: {decision.selected_worker_species_id}")
# -> "wrk.doc.summarizer"

pyhall.models

All models are Pydantic v2 BaseModel classes.

Type Aliases

Env = Literal["dev", "stage", "prod", "edge"]
DataLabel = Literal["PUBLIC", "INTERNAL", "RESTRICTED"]
QoSClass = Literal["P0", "P1", "P2", "P3"]
TenantRisk = Literal["low", "medium", "high"]

RouteInput

The capability request envelope sent to the Hall. Maps to WCP spec section 4.1.

FieldTypeRequiredDescription
capability_idstryesWCP capability ID, e.g. cap.doc.summarize
envEnvyesDeployment environment: dev | stage | prod | edge
data_labelDataLabelyesData sensitivity: PUBLIC | INTERNAL | RESTRICTED
tenant_riskTenantRiskyesTenant risk tier: low | medium | high
qos_classQoSClassyesQoS priority: P0 (highest) through P3 (background)
tenant_idstryesIdentifier of the requesting tenant. Must not be empty or whitespace.
correlation_idstryesUUID v4. Must be propagated through all downstream calls.
requestDict[str, Any]noArbitrary payload for the target worker. Default: {}
blast_radiusOptional[Dict[str, Any]]noPre-computed blast radius dimensions.
blast_scoreOptional[int]noPre-computed blast score (0–100). Router computes if None. Must not be bool.
privilege_contextOptional[Dict[str, Any]]noPrivilege context for envelope enforcement.
dry_runboolnoIf True, decision is made but no worker is dispatched. Default: False
tenant_credentialOptional[str]noSigned WCP credential. Optional in v0.1; required when HallConfig.require_credential=True (v0.2+).

RouteDecision

The routing decision returned by the Hall. On success: denied=False, selected_worker_species_id is set. On denial: denied=True, deny_reason_if_denied is set.

FieldTypeDescription
decision_idstrUUID v4 identifying this routing decision.
timestampstrISO 8601 UTC timestamp of the decision.
correlation_idstrPropagated from RouteInput.correlation_id.
tenant_idstrPropagated from RouteInput.tenant_id.
capability_idstrThe capability that was requested.
matched_rule_idstrThe routing rule that matched. "NO_MATCH" if none matched.
envEnvPropagated from input.
data_labelDataLabelPropagated from input.
tenant_riskTenantRiskPropagated from input.
qos_classQoSClassPropagated from input.
deniedboolTrue if the request was denied.
deny_reason_if_deniedOptional[Dict[str, Any]]Denial reason dict. None on allow.
selected_worker_species_idOptional[str]The dispatched worker species. None on denial.
candidate_workers_rankedList[CandidateWorker]All candidates considered, in rank order.
required_controls_effectiveList[str]Controls that were enforced.
recommended_profiles_effectiveList[Dict[str, Any]]Recommended profiles from the matched rule.
escalation_effectiveEscalationEscalation policy applied.
preconditions_checkedPreconditionsCheckedPrecondition flags.
artifact_hashOptional[str]SHA-256 of the serialized RouteInput. Proves what was routed.
worker_attestation_checkedboolTrue when the Hall verified the worker’s code hash.
worker_attestation_validOptional[bool]True = hash matched. False = TAMPERED. None = not checked.
dry_runboolTrue when this was a dry-run decision.
telemetry_envelopesList[Dict[str, Any]]Mandatory WCP telemetry events (always 3+).

CandidateWorker

A worker species candidate considered during routing.

FieldTypeDescription
worker_species_idstrWCP worker species ID, e.g. wrk.doc.summarizer.
score_hintOptional[float]Pre-ranked score from the rules engine.
requires_controls_minimumOptional[List[str]]Minimum controls this candidate requires.
skip_reasonOptional[str]Why this candidate was not selected.

Escalation

Escalation policy from the matched routing rule.

FieldTypeDefaultDescription
policy_gateboolFalseWhether the policy gate must be evaluated.
msavx_step_upboolFalseWhether MSAVX step-up approval is required.
human_required_defaultboolFalseWhether human review is required by default.
human_required_ifList[Dict[str, Any]][]Conditional human review triggers.
rationaleOptional[str]NoneReason for escalation requirement.

HallConfig

Hall-level governance configuration. Sets floors that individual routing rules cannot lower.

FieldTypeDefaultDescription
require_signatoryboolFalseDeny requests from tenants not in allowed_tenants.
allowed_tenantsList[str][]Registered signatory tenant IDs.
require_credentialboolFalseNot enforced in v0.1. Planned for v0.2.
require_worker_attestationboolFalseVerify worker code hash before dispatch (WCP §5.10).
enforce_correlation_idboolTrueRequire correlation_id on every request.
enforce_required_controlsboolTrueEnforce controls check regardless of rule.
enforce_blast_scoring_in_prodboolTrueEnforce blast scoring in prod/edge.
enforce_privilege_envelopesboolFalsePrivilege envelope enforcement (opt-in per rule in v0.1).

PreconditionsChecked

Precondition flags applied during routing.

FieldTypeDefault
must_have_correlation_idboolTrue
must_attach_policy_versionboolTrue
must_record_artifact_hash_if_executesboolTrue
deny_if_missing_required_controlsboolTrue
deny_if_unsigned_artifact_in_prodboolFalse
deny_if_no_attestation_in_prodboolFalse

pyhall.router

make_decision()

The Hall’s core routing function. Never raises in normal operation — all failures are expressed as denied RouteDecision objects.

from pyhall.router import make_decision
def make_decision(
inp: RouteInput,
rules: List[Rule],
registry_controls_present: set[str],
registry_worker_available: Callable[[str], bool],
registry_get_privilege_envelope: Optional[Callable[[str], Optional[dict]]] = None,
registry_policy_allows_privilege: Optional[Callable[[str, str, dict], tuple[bool, str]]] = None,
policy_gate_eval: Optional[Callable[[dict], tuple[str, str, str]]] = None,
conformance_spec: Optional[Dict[str, Any]] = None,
task_id: str = "task_default",
hall_config: Optional[HallConfig] = None,
registry_get_worker_hash: Optional[Callable[[str], Optional[str]]] = None,
get_current_worker_hash: Optional[Callable[[str], Optional[str]]] = None,
) -> RouteDecision:

Parameters:

ParameterTypeRequiredDescription
inpRouteInputyesThe capability request.
rulesList[Rule]yesOrdered routing rules from load_rules().
registry_controls_presentset[str]yesControls declared present. From Registry.controls_present().
registry_worker_availableCallable[[str], bool]yesReturns True if the species is enrolled. From Registry.worker_available.
registry_get_privilege_envelopeOptional[Callable]noReturns privilege envelope for a species.
registry_policy_allows_privilegeOptional[Callable]noReturns (allowed, reason) for privilege enforcement.
policy_gate_evalOptional[Callable]noRequired when a matched rule has escalation.policy_gate=True. Returns (decision, policy_version, reason) where decision is "ALLOW", "DENY", or "REQUIRE_HUMAN".
conformance_specOptional[Dict]noConformance spec for CI/test mode only. Raises RuntimeError on failure. Do not use in production.
task_idstrnoIdentifier for this routing task. Included in telemetry. Default: "task_default".
hall_configOptional[HallConfig]noHall-level governance floors.
registry_get_worker_hashOptional[Callable]noReturns registered code hash for a species. Required when hall_config.require_worker_attestation=True.
get_current_worker_hashOptional[Callable]noReturns current live code hash for a species. Required when hall_config.require_worker_attestation=True.

Returns: RouteDecision

Routing pipeline (WCP-Full):

  1. Match first routing rule (fail-closed on no match)
  2. Verify preconditions (correlation_id, etc.)
  3. Verify required controls against registry
  4. Blast radius scoring and gating
  5. Policy gate evaluation
  6. Select first available worker candidate
  7. Emit mandatory telemetry
  8. Optional conformance check
  9. Return RouteDecision

pyhall.registry

Registry

The Hall’s source of truth for enrolled workers. Loads enrollment records from a directory of JSON files.

from pyhall.registry import Registry
# Load from directory
registry = Registry(registry_dir="/path/to/enrolled/")
# Or build programmatically
registry = Registry()
registry.enroll(record_dict)

Constructor:

Registry(registry_dir: Optional[str] = None)

Enrollment:

MethodSignatureDescription
enroll(record: Dict[str, Any]) -> NoneEnroll a single worker from a registry record dict.

Controls:

MethodSignatureDescription
controls_present() -> set[str]Return the set of currently declared controls.
set_controls_present(controls: List[str]) -> NoneOverride the full set of present controls.
add_controls_present(controls: List[str]) -> NoneAdd controls to the existing set.

Worker availability:

MethodSignatureDescription
worker_available(worker_species_id: str) -> boolReturn True if the species is enrolled and available.
workers_for_capability(capability_id: str) -> List[str]Return all enrolled species that handle this capability.
set_workers_available(worker_species_ids: List[str]) -> NoneOverride the full set of available species.
add_workers_available(worker_species_ids: List[str]) -> NoneMark additional species as available.

Privilege envelopes:

MethodSignatureDescription
get_privilege_envelope(worker_species_id: str) -> Optional[dict]Return the privilege envelope for a species.
set_privilege_envelopes(envelopes: Dict[str, dict]) -> NoneMap worker_species_id -> privilege envelope.
set_egress_allowlist(env: str, allowlist: List[str]) -> NoneConfigure the egress allowlist for an environment.
policy_allows_privilege(env: str, data_label: str, envelope: Optional[dict]) -> Tuple[bool, str]Evaluate whether the privilege envelope is permitted. Returns (allowed, reason).

Worker code attestation (WCP §5.10):

MethodSignatureDescription
register_attestation(species_id: str, source_file: str) -> strHash the worker’s source file and register as known-good. Returns SHA-256 hex digest.
get_worker_hash(species_id: str) -> Optional[str]Return the registered (known-good) SHA-256 hash.
compute_current_hash(species_id: str) -> Optional[str]Read the worker source file from disk and compute its current hash.
attestation_callables() -> Tuple[Callable, Callable]Return (get_worker_hash, compute_current_hash) for use in make_decision().

Path allowlist:

MethodSignatureDescription
set_allowed_worker_dirs(dirs: List[str]) -> NoneRestrict worker source_file paths to these directories. Prevents path traversal.

Introspection:

MethodSignatureDescription
enrolled_count() -> intReturn the number of enrolled workers.
enrolled_workers() -> List[dict]Return all enrolled worker records.
summary() -> dictReturn a status summary dict for display or health checks.

Worker registry record format:

{
"worker_id": "org.example.my-summarizer",
"worker_species_id": "wrk.doc.summarizer",
"capabilities": ["cap.doc.summarize"],
"risk_tier": "low",
"required_controls": ["ctrl.obs.audit-log-append-only"],
"currently_implements": ["ctrl.obs.audit-log-append-only"],
"allowed_environments": ["dev", "stage", "prod"],
"blast_radius": {"data": 1, "network": 0, "financial": 0, "time": 1},
"privilege_envelope": {
"secrets_access": [],
"network_egress": "none",
"filesystem_writes": ["/tmp/"],
"tools": []
},
"owner": "org.example",
"contact": "team@example.com"
}

pyhall.rules

Rule

A single WCP routing rule (frozen dataclass).

from pyhall.rules import Rule
@dataclass(frozen=True)
class Rule:
rule_id: str # Unique rule identifier, e.g. "rr_doc_summarize_dev_001"
match: Dict[str, Any] # Match conditions
decision: Dict[str, Any] # Decision payload

Match conditions support:

  • Exact: "data_label": "INTERNAL"
  • Membership: "env": {"in": ["dev", "stage"]}
  • Wildcard: "capability_id": {"any": true}

load_rules()

from pyhall.rules import load_rules
def load_rules(seed_path: str | Path) -> List[Rule]:

Load routing rules from a JSON file. The file must have a top-level "rules" array.

Raises FileNotFoundError, KeyError (missing "rules" key), or json.JSONDecodeError.

load_rules_from_dict()

def load_rules_from_dict(doc: Dict[str, Any]) -> List[Rule]:

Load routing rules from an already-parsed dict. Useful for testing.

route_first_match()

def route_first_match(rules: List[Rule], inp: Dict[str, Any]) -> Optional[Rule]:

Return the first rule that matches inp, or None. Per WCP §5.1, None MUST produce a denied routing decision.

Example rules.json:

{
"rules": [
{
"rule_id": "rr_doc_summarize_dev_001",
"match": {
"capability_id": "cap.doc.summarize",
"env": {"in": ["dev", "stage"]},
"data_label": "INTERNAL"
},
"decision": {
"candidate_workers_ranked": [
{"worker_species_id": "wrk.doc.summarizer", "score_hint": 1.0}
],
"required_controls_suggested": ["ctrl.obs.audit-log-append-only"],
"escalation": {"policy_gate": false, "human_required_default": false},
"preconditions": {}
}
}
]
}

pyhall.policy_gate

PolicyGate

Evaluates a capability request context and returns a (decision, policy_version, reason) triple. The default implementation allows everything. Subclass to implement real governance.

from pyhall.policy_gate import PolicyGate
class PolicyGate:
def evaluate(self, context: Dict[str, Any]) -> Tuple[str, str, str]:
...

evaluate() parameters:

context dict contains:

  • capability_id
  • tenant_id
  • env
  • data_label
  • tenant_risk
  • qos_class
  • policy_version

Returns: Tuple[str, str, str](decision, policy_version, reason)

  • decision: "ALLOW" | "DENY" | "REQUIRE_HUMAN"
  • policy_version: the policy version string evaluated
  • reason: human-readable reason string

Custom policy gate example:

class MyPolicyGate(PolicyGate):
def evaluate(self, context):
if context["env"] == "prod" and context["data_label"] == "RESTRICTED":
return ("REQUIRE_HUMAN", "policy.v1", "restricted_data_in_prod")
return ("ALLOW", "policy.v1", "default_allow")
gate = MyPolicyGate()
decision = make_decision(inp, rules, ..., policy_gate_eval=gate.evaluate)

Hall API Server

The Hall API is an HTTP interface to the Hall registry and routing decision log. Each organization runs their own instance.

Start the server:

Terminal window
pip install pyhall-wcp flask
python -m hall_api.server

Environment variables:

VariableDefaultDescription
HALL_API_PORT8765Port to listen on.
HALL_API_HOST127.0.0.1Host to bind.
HALL_DB_PATHhall.dbPath to the SQLite database.

CLI

The pyhall CLI ships with pip install pyhall-wcp.

Terminal window
# Route a capability request
pyhall route --capability cap.doc.summarize --env dev --tenant-id demo
# Validate all test fixtures against routing rules
pyhall validate --rules rules.json --tests tests.json
# Show registry status
pyhall status --registry-dir enrolled/
# Enroll a worker
pyhall enroll my_worker/registry_record.json
# Interactive wizard to scaffold a complete WCP worker package
pyhall build
# Fuzzy search across taxonomy catalog entities
pyhall search "summarize documents"
# Detailed lookup for a specific catalog entity
pyhall explain cap.doc.summarize
# Browse the taxonomy catalog (interactive)
pyhall browse
pyhall browse --type cap
# Simulate routing a capability request
pyhall dispatch cap.doc.summarize --env prod
# Show PyHall and WCP spec versions
pyhall version

pyhall.registry_client

RegistryClient

HTTP client for the pyhall.dev registry API. Zero external dependencies — stdlib urllib only.

from pyhall import RegistryClient, RegistryRateLimitError
client = RegistryClient() # default: https://api.pyhall.dev
r = client.verify("org.example.my-worker")
print(r.status) # 'active' | 'revoked' | 'banned' | 'unknown'

Constructor:

RegistryClient(
base_url: Optional[str] = None, # override via PYHALL_REGISTRY_URL env var
session_token: Optional[str] = None, # pyhall_session JWT for authenticated calls
bearer_token: Optional[str] = None, # JWT bearer token for authenticated calls
timeout: int = 10, # request timeout in seconds
cache_ttl: float = 60.0, # verify() cache TTL in seconds
)

Public endpoints (no auth required):

MethodSignatureDescription
verify(worker_id: str) -> VerifyResponseGet a worker’s attestation status. Returns status='unknown' on 404 (IDOR-safe).
is_hash_banned(sha256: str) -> boolReturns True if the hash appears in confirmed ban-list entries.
get_ban_list(limit: int = 500) -> list[BanEntry]Fetch all confirmed ban-list entries.
health() -> dictReturns {'ok': bool, 'version': str}.
prefetch(worker_ids: list[str]) -> NonePre-populate the verify cache. Non-fatal on 404 or rate limit.
get_worker_hash(worker_id: str) -> Optional[str]Returns current_hash for active workers; None otherwise. Suitable as registry_get_worker_hash callback in make_decision().

Authenticated endpoints (require session_token or bearer_token):

MethodSignatureDescription
report_hash(sha256: str, reason: str, evidence_url: Optional[str]) -> NoneSubmit a community hash report. Written as confirmed=0 pending admin review.
submit_attestation(worker_id, package_hash, *, label, ai_generated, ai_service, ai_model, ai_session_id, bearer_token) -> AttestationResponseSubmit a full-package attestation to the registry.

Data models:

@dataclass
class VerifyResponse:
worker_id: str
status: str # 'active' | 'revoked' | 'banned' | 'unknown'
current_hash: Optional[str]
banned: bool
ban_reason: Optional[str]
attested_at: Optional[str]
ai_generated: bool
ai_service: Optional[str]
ai_model: Optional[str]
ai_session_fingerprint: Optional[str]
@dataclass
class BanEntry:
sha256: str
reason: str
reported_at: str
source: str
review_status: Optional[str] = None
@dataclass
class AttestationResponse:
id: str
worker_id: str
sha256: str

Registry + make_decision() integration:

from pyhall import make_decision, RegistryClient, Registry, load_rules
client = RegistryClient()
# Pre-populate cache so the synchronous callback has data
client.prefetch(["org.example.my-worker"])
rules = load_rules("rules.json")
registry = Registry(registry_dir="enrolled/")
decision = make_decision(
inp=inp,
rules=rules,
registry_controls_present=registry.controls_present(),
registry_worker_available=registry.worker_available,
registry_get_worker_hash=client.get_worker_hash,
)

pyhall.attestation

Full-package attestation for WCP workers. The unit of attestation is the complete worker package directory — code, dependencies, and config all hashed together.

Signing model: HMAC-SHA256. Signing secret is read from the WCP_ATTEST_HMAC_KEY environment variable.

Deny codes (fail-closed — no silent fallback):

ConstantMeaning
ATTEST_MANIFEST_MISSINGmanifest.json absent or unreadable
ATTEST_MANIFEST_ID_MISMATCHManifest identity doesn’t match declared worker
ATTEST_HASH_MISMATCHRecomputed package hash doesn’t match manifest
ATTEST_SIGNATURE_MISSINGNo signature in manifest or WCP_ATTEST_HMAC_KEY not set
ATTEST_SIG_INVALIDHMAC-SHA256 signature verification failed
ATTEST_BANNED_HASHPackage hash matches a banned hash in the registry’s banned_hashes set

scaffold_package()

Create a minimal worker package directory layout.

from pathlib import Path
from pyhall import scaffold_package
scaffold_package(
package_root=Path("my-worker/"),
worker_logic_file=Path("my_logic.py"), # optional; writes stub if None
overwrite=False,
)
# Creates:
# my-worker/code/bootstrap.py
# my-worker/code/worker_logic.py
# my-worker/requirements.lock
# my-worker/config.schema.json

canonical_package_hash()

Compute a deterministic SHA-256 hash over all files in a worker package directory. Excludes manifest.json, .git/, __pycache__/, .pyc files.

from pathlib import Path
from pyhall import canonical_package_hash
pkg_hash = canonical_package_hash(Path("my-worker/"))
# → 64-character lowercase hex SHA-256

build_manifest()

Build and HMAC-sign a worker package manifest. Does not write to disk.

from pathlib import Path
from pyhall import build_manifest
manifest = build_manifest(
package_root=Path("my-worker/"),
worker_id="org.example.my-worker.instance-1",
worker_species_id="wrk.example.my-worker",
worker_version="1.0.0",
signing_secret="your-secret", # or set WCP_ATTEST_HMAC_KEY
build_source="local", # 'local' | 'ci' | 'agent'
)
# Returns a signed manifest dict with 'signature_hmac_sha256' field

write_manifest()

Write a signed manifest dict to disk as formatted JSON.

from pathlib import Path
from pyhall import write_manifest
write_manifest(manifest, Path("my-worker/manifest.json"))

PackageAttestationVerifier

Verifies that a worker package is attested and unchanged at runtime. Fail-closed — any mismatch returns a deny code and ok=False.

from pathlib import Path
from pyhall import PackageAttestationVerifier
verifier = PackageAttestationVerifier(
package_root=Path("/opt/workers/my-worker"),
manifest_path=Path("/opt/workers/my-worker/manifest.json"),
worker_id="org.example.my-worker.instance-1",
worker_species_id="wrk.example.my-worker",
# secret_env defaults to "WCP_ATTEST_HMAC_KEY"
)
ok, deny_code, meta = verifier.verify()
if not ok:
raise SystemExit(f"Attestation denied: {deny_code}")
# meta["trust_statement"] — canonical namespace-key trust claim
# meta["package_hash"] — verified hash for evidence receipts
# meta["verified_at_utc"] — UTC ISO 8601

Worker package workflow

from pathlib import Path
from pyhall import scaffold_package, build_manifest, write_manifest, PackageAttestationVerifier
import os
pkg = Path("my-worker/")
# 1. Scaffold the package layout
scaffold_package(pkg, worker_logic_file=Path("my_logic.py"))
# 2. Build and sign the manifest (at build/CI time)
manifest = build_manifest(
package_root=pkg,
worker_id="org.example.my-worker.instance-1",
worker_species_id="wrk.example.my-worker",
worker_version="1.0.0",
signing_secret=os.environ["WCP_ATTEST_HMAC_KEY"],
build_source="ci",
)
# 3. Write manifest alongside the package
write_manifest(manifest, pkg / "manifest.json")
# 4. At runtime — verify before execution
verifier = PackageAttestationVerifier(
package_root=pkg,
manifest_path=pkg / "manifest.json",
worker_id="org.example.my-worker.instance-1",
worker_species_id="wrk.example.my-worker",
)
ok, deny_code, meta = verifier.verify()
if not ok:
raise SystemExit(f"Worker tampered: {deny_code}")

Additional exports

detect_shadow_rules()

Detect rules that may shadow later, more specific rules in a rule list.

from pyhall import detect_shadow_rules, load_rules
rules = load_rules("rules.json")
shadows = detect_shadow_rules(rules)
for s in shadows:
print(f"Rule {s['shadowing_rule_id']} shadows {s['shadowed_rule_id']}")

A rule A shadows rule B when A appears before B and A’s match conditions are a semantic superset of B’s — meaning B never fires. Useful for validating rule sets in CI.


Public exports summary

from pyhall import (
# Core routing
make_decision,
detect_shadow_rules,
# Models
RouteInput,
RouteDecision,
HallConfig,
# Registry
Registry,
PolicyGate,
# Rules
Rule,
load_rules,
load_rules_from_dict,
# Registry client
RegistryClient,
RegistryRateLimitError,
VerifyResponse,
BanEntry,
AttestationResponse,
# Package attestation
PackageAttestationVerifier,
canonical_package_hash,
build_manifest,
write_manifest,
scaffold_package,
ATTEST_MANIFEST_MISSING,
ATTEST_MANIFEST_ID_MISMATCH,
ATTEST_HASH_MISMATCH,
ATTEST_SIGNATURE_MISSING,
ATTEST_SIG_INVALID,
ATTEST_BANNED_HASH,
)