AWS Strands

HighflameStrandsHooks for AWS Strands — HookProvider with BeforeInvocationEvent, AfterModelCallEvent, BeforeToolCallEvent, AfterToolCallEvent. Available for Python and TypeScript. event.cancel_tool se

HighflameStrandsHooks integrates Highflame Shield into AWS Strands Agents by implementing the Strands HookProvider interface. It intercepts agent invocations, model responses, and tool calls without modifying agent or tool definitions.

Available for Python and TypeScript.

Installation

Python

pip install 'highflame[strands]'

TypeScript

Basic Usage

Python

from highflame import Highflame
from highflame.integrations.strands import HighflameStrandsHooks
from strands import Agent

client = Highflame(api_key="hf_sk_...")
hooks = HighflameStrandsHooks(client, mode="enforce")

agent = Agent(
    model=model,
    tools=[my_tool],
    hooks=[hooks],
)

result = await agent.invoke_async("Your prompt here")

TypeScript

On a policy violation, HighflameStrandsHooks raises BlockedError. For tool call denials, it also sets event.cancel_tool to the deny reason before raising, following the idiomatic Strands cancellation pattern.

Constructor

Python

HighflameStrandsHooks(
    client: Highflame,
    *,
    mode: str = "enforce",
    session_id: str | None = None,
    session_id_key: str = "session_id",
)

Parameter	Type	Default	Description
client	Highflame	required	Initialized Highflame client
mode	str	"enforce"	Enforcement mode: "enforce", "monitor", or "alert"
session_id	str | None	None	Static session ID. If set, takes priority over invocation state.
session_id_key	str	"session_id"	Key to read from event.invocation_state for dynamic session ID.

TypeScript

Session ID Resolution

The hooks resolve the session ID in this order:

1. Static session_id / sessionId — if explicitly provided at construction time

2. event.invocation_state[session_id_key] / event.invocation_state[sessionIdKey] — from the Strands invocation state

3. None / undefined — no session tracking

To pass a session ID dynamically:

Python

result = await agent.invoke_async(
    "Analyze this document",
    invocation_state={"session_id": f"user-{user_id}"},
)

TypeScript

Hooks

HighflameStrandsHooks registers four async hooks with the Strands hook registry:

BeforeInvocationEvent

Fires before the agent loop starts for a new invocation. Extracts the last user message from event.messages and evaluates it as a prompt.

• Content type: "prompt", action: "process_prompt"

• On deny: raises BlockedError before the agent processes the request

AfterModelCallEvent

Fires after the LLM produces a response. Extracts text from event.stop_response.message.content and evaluates it as a model response.

• Content type: "response", action: "process_prompt"

• On deny: raises BlockedError before the response flows back into the agent loop

BeforeToolCallEvent

Fires before a tool executes. Evaluates the tool call using event.tool_use.name and event.tool_use.input.

• Content type: "tool_call", action: "call_tool"

• On deny: sets event.cancel_tool to the deny reason, then raises BlockedError

Setting event.cancel_tool is the idiomatic Strands way to signal tool cancellation. It is set before raising to ensure Strands runtime state is consistent even if the exception is caught upstream.

AfterToolCallEvent

Fires after a tool returns. Extracts text from event.result and evaluates it as a tool response.

• Content type: "response", action: "call_tool"

• On deny: raises BlockedError, result is not returned to the agent

Registration

Python

Strands HookProvider objects register themselves through register_hooks(registry), which Strands calls automatically when you pass the provider to the Agent constructor:

# Strands calls register_hooks automatically
agent = Agent(model=model, tools=[...], hooks=[hooks])

TypeScript

Enforcement Modes

Python

hooks = HighflameStrandsHooks(client, mode="enforce")  # block violations (default)
hooks = HighflameStrandsHooks(client, mode="monitor")  # observe without blocking
hooks = HighflameStrandsHooks(client, mode="alert")    # allow but trigger alerting

TypeScript

Complete Example

Python

import asyncio
from strands import Agent
from strands.models.bedrock import BedrockModel

from highflame import Highflame, BlockedError
from highflame.integrations.strands import HighflameStrandsHooks

client = Highflame(api_key="hf_sk_...")
hooks = HighflameStrandsHooks(client, mode="enforce", session_id_key="session_id")

model = BedrockModel(model_id="us.anthropic.claude-sonnet-4-5-20251101-v2:0")

agent = Agent(model=model, tools=[file_tool, web_tool], hooks=[hooks])

async def handle_request(user_input: str, session_id: str):
    try:
        return await agent.invoke_async(
            user_input,
            invocation_state={"session_id": session_id},
        )
    except BlockedError as e:
        return {"error": f"Request blocked: {e.response.policy_reason}"}

asyncio.run(handle_request("Analyze the quarterly report", "user-42-session-7"))

TypeScript

Content Extraction

The integration handles both Bedrock-native content formats:

• Plain strings

• [{"type": "text", "text": "..."}] — standard content block format

• [{"text": "..."}] — Bedrock native format (no "type" key)

• Objects with type/text attributes (Python SDK objects)

Error Handling

Python

from highflame import BlockedError

try:
    result = await agent.invoke_async(user_input)
except BlockedError as e:
    print(f"Blocked: {e.response.policy_reason}")
    print(f"Signals: {[s.name for s in e.response.signals]}")

TypeScript

Requirements

Python

Package	Minimum Version
strands-agents	1.0+
highflame	latest

TypeScript