Advanced Topics

Advanced Python Shield topics including request debugging, optimization, streaming, agentic context, sessions, and client configuration.

This page collects the deeper operational and debugging features of the Python SDK.

Advanced Request Options

Use explain, debug, optimize, dryrun, early_exit, and contexts for debugging, profiling, and performance tuning.

# explain=True — get the full policy trace
resp = client.guard.evaluate(GuardRequest(
    content=user_input,
    content_type="prompt",
    action="process_prompt",
    explain=True,
))
print(resp.explanation)
print(resp.root_causes)
print(resp.tiers_evaluated)
print(resp.projected_context)

# debug=True — get per-detector results (implies explain=True)
resp = client.guard.evaluate(GuardRequest(
    content=user_input,
    content_type="prompt",
    action="process_prompt",
    debug=True,
))
for d in resp.detectors:
    print(f"{d.name}: {d.context}")

# optimize=True — only run detectors referenced by active policies
resp = client.guard.evaluate(GuardRequest(
    content=user_input,
    content_type="prompt",
    action="process_prompt",
    optimize=True,
))
if resp.optimization:
    print(f"Ran {len(resp.optimization.required_detectors)} detectors")
    print(f"Skipped {len(resp.optimization.skipped_detectors)} detectors")

# dryrun=True — see which detectors would run without executing them
resp = client.guard.evaluate(GuardRequest(
    content=user_input,
    content_type="prompt",
    action="process_prompt",
    optimize=True,
    dryrun=True,
))

# early_exit=True — stop after the first tier that denies
resp = client.guard.evaluate(GuardRequest(
    content=user_input,
    content_type="prompt",
    action="process_prompt",
    early_exit=True,
))
print(resp.tiers_skipped)

# contexts — pass reference material for grounding and hallucination detection
resp = client.guard.evaluate(GuardRequest(
    content=model_response,
    content_type="response",
    action="process_prompt",
    contexts=[original_prompt, *rag_chunks],
))

Agentic Context

Pass typed context objects to provide richer signal to detectors and Cedar policies.

ToolContext

from highflame import GuardRequest, ToolContext

resp = client.guard.evaluate(GuardRequest(
    content="execute shell command",
    content_type="tool_call",
    action="call_tool",
    tool=ToolContext(
        name="shell",
        arguments={"cmd": "ls /etc", "timeout": 30},
        server_id="mcp-server-001",
        is_builtin=False,
    ),
))

Field	Type	Description
name	str	Tool name
arguments	dict[str, Any] | None	Tool arguments
server_id	str | None	MCP server that registered this tool
is_builtin	bool | None	Whether the tool is a first-party built-in
description	str | None	Tool description

ModelContext

from highflame import GuardRequest, ModelContext

resp = client.guard.evaluate(GuardRequest(
    content="user prompt",
    content_type="prompt",
    action="process_prompt",
    model=ModelContext(
        provider="anthropic",
        model="claude-sonnet-4-6",
        temperature=0.7,
        tokens_used=1500,
        max_tokens=4096,
    ),
))

Field	Type	Description
provider	str | None	Model provider
model	str | None	Model identifier
temperature	float | None	Sampling temperature
tokens_used	int | None	Tokens consumed this turn
max_tokens	int | None	Token limit for this turn

MCPContext and FileContext

from highflame import MCPContext, FileContext, GuardRequest

# MCP server connection
resp = client.guard.evaluate(GuardRequest(
    content="connect to MCP server",
    content_type="tool_call",
    action="connect_server",
    mcp=MCPContext(
        server_name="filesystem-server",
        server_url="http://mcp.internal:8080",
        transport="http",
        verified=False,
        capabilities=["read_file", "write_file", "shell"],
    ),
))

# File write
resp = client.guard.evaluate(GuardRequest(
    content="env vars and secrets here",
    content_type="file",
    action="write_file",
    file=FileContext(
        path="/app/.env",
        operation="write",
        size=512,
        mime_type="text/plain",
    ),
))

MCPContext fields:

Field	Type	Description
server_name	str	MCP server name
server_url	str | None	MCP server URL
transport	str | None	Transport protocol: "sse", "stdio", or "http"
verified	bool | None	Whether the server passed verification
capabilities	list[str] | None	Advertised capabilities from the server manifest

FileContext fields:

Field	Type	Description
path	str	File path
operation	str	File operation such as "read", "write", or "append"
size	int | None	File size in bytes
mime_type	str | None	File MIME type
file_name	str | None	Original file name
file_extension	str | None	File extension such as "pdf" or "docx"
sensitivity_level	str | None	"public", "internal", "confidential", or "restricted"
mip_label_id	str | None	Microsoft Information Protection label GUID
mip_label_name	str | None	MIP label display name
is_encrypted	bool | None	Whether the file is encrypted
is_rights_managed	bool | None	Whether the file has IRM or RMS restrictions

SSE Streaming

The streaming endpoint yields detection results as they arrive during the tiered evaluation pipeline.

from highflame import Highflame, GuardRequest

with Highflame(api_key="hf_sk_...") as client:
    for event in client.guard.stream(GuardRequest(
        content="execute sudo rm -rf /",
        content_type="tool_call",
        action="call_tool",
    )):
        if event.type == "decision":
            print(f"Final decision: {event.data.get('decision')}")

Async streaming:

async with Highflame(api_key="hf_sk_...") as client:
    async for event in client.guard.astream(GuardRequest(
        content="user prompt text",
        content_type="prompt",
        action="process_prompt",
    )):
        if event.type == "detection":
            print(f"Detector: {event.data.get('detector_name')}")
        elif event.type == "decision":
            print(f"Decision: {event.data.get('decision')}")

event.type	Description
"detection"	A detector tier completed
"decision"	Final allow or deny decision
"error"	Stream error
"done"	Stream ended

Error Handling

from highflame import (
    HighflameError,
    APIError,
    AuthenticationError,
    RateLimitError,
    APIConnectionError,
    BlockedError,
)

try:
    resp = client.guard.evaluate(request)
except BlockedError as e:
    print(f"Blocked: {e.response.policy_reason}")
except AuthenticationError as e:
    print(f"Auth failed: {e.detail}")
except RateLimitError as e:
    print(f"Rate limited: {e.detail}")
except APIError as e:
    print(f"API error {e.status}: {e.title} — {e.detail}")
except APIConnectionError as e:
    print(f"Could not reach service: {e}")
except HighflameError as e:
    print(f"Error: {e}")

Exception	When raised	Key attributes
BlockedError	Decorator receives decision == "deny"	response: GuardResponse
AuthenticationError	401 Unauthorized	status, title, detail
RateLimitError	429 Too Many Requests	status, title, detail
APIError	Non-2xx HTTP response from the service	status, title, detail
APIConnectionError	Timeout or network failure	—
HighflameError	Base class	—

Enforcement Modes

Mode	Behavior	resp.denied	resp.alerted
"enforce"	Block on deny	True on deny	False
"monitor"	Allow and log silently	False	False
"alert"	Allow and trigger alerting pipeline	False	True if violated

# Monitor — observe without blocking
resp = client.guard.evaluate(GuardRequest(
    content=user_input,
    content_type="prompt",
    action="process_prompt",
    mode="monitor",
))
if resp.actual_decision == "deny":
    shadow_log.record(user_input, resp.policy_reason)

# Alert — allow but signal the alerting pipeline
resp = client.guard.evaluate(GuardRequest(..., mode="alert"))
if resp.alerted:
    pagerduty.trigger(resp.policy_reason)

# Enforce — block violations (default)
resp = client.guard.evaluate(GuardRequest(..., mode="enforce"))
if resp.denied:
    raise PermissionError(f"Request blocked: {resp.policy_reason}")

Session Tracking

Pass the same session_id across all turns of a conversation to enable cumulative risk tracking.

SESSION_ID = f"sess_{user_id}_{conversation_id}"

resp = client.guard.evaluate(GuardRequest(
    content=turn.content,
    content_type=turn.content_type,
    action=turn.action,
    session_id=SESSION_ID,
))

if resp.session_delta:
    print(f"Turn {resp.session_delta.turn_count}, risk: {resp.session_delta.cumulative_risk:.2f}")

Multi-Project Support

Pass account_id and project_id to scope all requests to a specific project:

client = Highflame(
    api_key="hf_sk_...",
    account_id="acc_123",
    project_id="proj_456",
)

Client Options

client = Highflame(
    api_key="hf_sk_...",
    base_url="https://...",
    token_url="https://...",
    timeout=30.0,
    max_retries=2,
    account_id="acc_123",
    project_id="proj_456",
)

Option	Type	Default	Description
api_key	str	required	Service key (hf_sk_...) or raw JWT
base_url	str	SaaS endpoint	Guard service URL
token_url	str	SaaS token URL	Token exchange URL
timeout	float	30.0	Per-request timeout in seconds
max_retries	int	2	Retries on transient errors
account_id	str | None	None	Optional account ID
project_id	str | None	None	Optional project ID
default_headers	dict[str, str] | None	None	Custom headers sent with every request

Individual methods also accept a timeout keyword argument to override the client-level timeout for that call:

resp = client.guard.evaluate(request, timeout=60.0)

Internal Usage

Internal services that call Shield for non-guardrails products must set the X-Product header so Shield routes the request to the correct Cedar evaluator and policy set.

from highflame import Highflame

sentry_client = Highflame(
    api_key="hf_sk_...",
    default_headers={"X-Product": "sentry"},
)

overwatch_client = Highflame(
    api_key="hf_sk_...",
    default_headers={"X-Product": "overwatch"},
)

mcp_client = Highflame(
    api_key="hf_sk_...",
    default_headers={"X-Product": "mcp_gateway"},
)

When X-Product is not set, Shield defaults to "guardrails". External customers should not need to set this header.