Python API Reference
Complete reference for all public functions and classes in FORGE v2.0.0.
normalize_content()
The primary entry point for all FORGE normalization. Accepts content and a content type, runs the appropriate lane chain, and returns a ForgeResult.
Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
content |
str | dict |
Yes | The raw content to normalize. String for text-based types; dict for JSON and VIDEO_META. |
content_type |
str |
Yes | One of: "DIFF", "TEXT", "JSON", "PROMPT", "COMPLIANCE", "SPEECH", "VIDEO_META". |
config |
ForgeConfig | None |
No | Custom configuration. If None, loads from env vars / YAML automatically. |
actor |
dict | None |
No | Actor metadata override. If None, uses config actor settings. |
Returns
Returns a ForgeResult instance with the normalized content, audit trail, lane results, and ForgeStamp.
Raises
| Exception | When |
|---|---|
ForgeConfigError |
Invalid configuration (missing HMAC secret in strict mode, unknown content type). |
ForgeTimeoutError |
Total normalization time exceeds timeout_ms. |
ForgeContentError |
Content is None, empty, or of an unsupported type for the declared content_type. |
from forge import normalize_content
# Simple call
result = normalize_content("Hello world", "TEXT")
# With custom config
from forge.config import ForgeConfig
config = ForgeConfig(max_iterations=20, strict_mode=True)
result = normalize_content(data, "JSON", config=config)
# With actor override
result = normalize_content(content, "DIFF", actor={
"service": "code-review-bot",
"version": "2.1.0",
"environment": "production"
})
ForgeResult
Returned by normalize_content(). Contains all outputs of the normalization pipeline.
Attributes
| Attribute | Type | Description |
|---|---|---|
normalized_content |
str | dict |
The cleaned, repaired output. Same type as the input content. |
audit_trail |
list[AuditEntry] |
Ordered list of every lane execution with timing and status. |
lane_results |
dict[str, LaneResult] |
Per-lane outcome keyed by lane ID (e.g., "T0", "L0.5"). |
content_type |
str |
The content type that was processed. |
passed |
bool |
True if trust level is TRUSTED or REPAIRED. |
stamp |
ForgeStamp |
The cryptographic attestation for this normalization run. |
iterations |
int |
Number of fixpoint loop iterations that ran. |
converged |
bool |
Whether the fixpoint loop converged (output stabilized). |
total_duration_ms |
float |
Total wall-clock time for the normalization call. |
Methods
Serialize the entire result to a JSON-compatible dictionary.
Serialize to a JSON string.
Human-readable summary: trust level, lanes run, iteration count, timing.
Pythonresult = normalize_content(content, "TEXT")
# Quick summary
print(result.summary())
# "TEXT | REPAIRED | 3 iterations | 4 lanes | 12.3ms"
# Full serialization
data = result.to_dict()
json_str = result.to_json(indent=2)
# Check convergence
if not result.converged:
print(f"Did not converge after {result.iterations} iterations")
ForgeStamp
Cryptographic attestation sealed with HMAC-SHA256. Attached to every ForgeResult as result.stamp.
Attributes
| Attribute | Type | Description |
|---|---|---|
trust_level |
str |
One of: "TRUSTED", "REPAIRED", "QUARANTINED", "REJECTED". |
lanes |
list[str] |
Lane IDs that executed during normalization. |
actor |
dict |
Actor information: service name, version, environment, instance ID. |
signature |
str |
HMAC-SHA256 hex digest sealing the stamp contents. |
timestamp |
str |
ISO 8601 timestamp of when the stamp was created. |
content_hash |
str |
SHA-256 hash of the normalized content (if include_content_hash is enabled). |
forge_version |
str |
FORGE library version that produced this stamp (e.g., "2.0.0"). |
Methods
Verify the HMAC-SHA256 signature. Uses the provided secret or falls back to FORGE_HMAC_SECRET.
Serialize the stamp to a JSON-compatible dictionary.
Serialize to a compact string suitable for HTTP headers (X-Forge-Stamp).
Class method. Deserialize a stamp from its header representation.
Pythonstamp = result.stamp
# Verify signature
assert stamp.verify() # Uses FORGE_HMAC_SECRET from env
assert stamp.verify(secret="my-custom-secret")
# Use in HTTP headers
response.headers["X-Forge-Stamp"] = stamp.to_header()
# Reconstruct from header
received_stamp = ForgeStamp.from_header(request.headers["X-Forge-Stamp"])
if received_stamp.verify():
print(f"Trust level: {received_stamp.trust_level}")
else:
print("Stamp verification failed — content may be tampered")
ForgeConfig
Configuration container. Loads from constructor args, environment variables, and YAML file.
Constructor Parameters
All configuration fields are accepted as keyword arguments. For example:
config = ForgeConfig(
hmac_secret="secret",
max_iterations=10,
telemetry_enabled=True,
strict_mode=False,
log_level="INFO"
)
Methods
Check if a specific lane is enabled.
Get lane-specific configuration dict.
Check if a content type is enabled.
Get max iterations for a content type (falls back to core setting).
Override a config value at runtime.
Override a lane-specific config value at runtime.
Validate the configuration. Returns a list of error strings (empty if valid).
Serialize the resolved configuration to a dictionary (secrets are masked).
AuditEntry
A single entry in the audit trail, representing one lane execution.
| Attribute | Type | Description |
|---|---|---|
lane_id |
str |
Lane identifier (e.g., "T0", "L0.5"). |
iteration |
int |
Which loop iteration this ran in (0 for pre-loop lanes). |
status |
str |
"PASSED", "REPAIRED", "WARNING", "ERROR". |
duration_ms |
float |
Execution time for this lane in milliseconds. |
changes_made |
int |
Number of repairs/modifications made. |
issues |
list[str] |
Human-readable descriptions of detected issues. |
LaneResult
Aggregate result for a single lane across all iterations.
| Attribute | Type | Description |
|---|---|---|
lane_id |
str |
Lane identifier. |
status |
str |
Final aggregate status across all iterations. |
total_changes |
int |
Total changes made across all iterations. |
total_duration_ms |
float |
Total time spent in this lane across all iterations. |
issues |
list[str] |
All issues detected across all iterations (deduplicated). |
iterations_run |
int |
Number of times this lane executed. |
Exceptions
Base class for all FORGE exceptions. All specific exceptions inherit from this.
Raised when configuration is invalid: missing required fields, unknown content types, invalid lane IDs.
Raised when total normalization time exceeds timeout_ms. The partial result is available via error.partial_result.
Raised when input content is None, empty, or the wrong type for the declared content_type (e.g., passing a string for "JSON" that isn't valid JSON).
Raised when a lane encounters an unrecoverable internal error. Includes error.lane_id and error.partial_result.
from forge import normalize_content
from forge.exceptions import (
ForgeError,
ForgeConfigError,
ForgeTimeoutError,
ForgeContentError,
ForgeLaneError
)
try:
result = normalize_content(content, "TEXT")
except ForgeTimeoutError as e:
print(f"Timed out after {e.elapsed_ms}ms")
# Partial result may be available
if e.partial_result:
print(f"Partial: {e.partial_result.summary()}")
except ForgeContentError as e:
print(f"Bad content: {e}")
except ForgeLaneError as e:
print(f"Lane {e.lane_id} failed: {e}")
except ForgeError as e:
print(f"FORGE error: {e}")
Module Structure
# Main entry point
from forge import normalize_content
# Configuration
from forge.config import ForgeConfig
# Result types
from forge.result import ForgeResult, ForgeStamp, AuditEntry, LaneResult
# Exceptions
from forge.exceptions import (
ForgeError,
ForgeConfigError,
ForgeTimeoutError,
ForgeContentError,
ForgeLaneError
)
# Content type constants
from forge.types import (
DIFF, TEXT, JSON, PROMPT,
COMPLIANCE, SPEECH, VIDEO_META
)
# Trust level constants
from forge.types import (
TRUSTED, REPAIRED, QUARANTINED, REJECTED
)
# Version
from forge import __version__ # "2.0.0"