# Advanced Topics

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.

```python
# 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`

```python
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`

```python
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`

```python
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.

```python
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:**

```python
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

```python
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 |

```python
# 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.

```python
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:

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

## Client Options

```python
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:

```python
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.

```python
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.


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://docs.highflame.ai/api-reference/sdk/shield/python-sdk/advanced-topics.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.