CrewAI

HighflameCrewHooks for CrewAI — register/unregister pattern, hooks on before/after LLM and tool calls. Available for Python and TypeScript. Session ID from crew.id automatically.

HighflameCrewHooks integrates Highflame Shield into CrewAI agents by registering with CrewAI's execution hooks registry. It intercepts every LLM call and tool execution without modifying task or agent definitions.

Available for Python and TypeScript.

Installation

Python

pip install 'highflame[crewai]'

TypeScript

Basic Usage

Python

from highflame import Highflame
from highflame.integrations.crewai import HighflameCrewHooks

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

result = crew.kickoff()

TypeScript

On a policy violation, HighflameCrewHooks raises BlockedError before the blocked operation completes.

Constructor

Python

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

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 not set, uses context.crew.id automatically.

TypeScript

Session ID Resolution

The hooks resolve the session ID in this order:

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

2. context.crew.id (Python) / crew[sessionIdKey] (TypeScript) — automatically derived from the running crew

3. None / undefined — no session tracking

Using the crew ID as the session ID is the right default for per-crew risk tracking. Pass an explicit session_id when you need to correlate across multiple crew runs.

Hooks

HighflameCrewHooks registers four hooks with the CrewAI hooks registry:

before_llm_call / beforeLlmCall

Fires before the LLM is called. Extracts the last user message from the message list and evaluates it as a prompt.

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

• On deny: raises BlockedError before the LLM is invoked

after_llm_call / afterLlmCall

Fires after the LLM produces a response. Evaluates context.response as a model response.

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

• On deny: raises BlockedError before the response is returned

before_tool_call / beforeToolCall

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

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

• On deny: raises BlockedError, tool is not called

after_tool_call / afterToolCall

Fires after a tool returns. Evaluates the tool result as a tool response.

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

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

Registration Patterns

Python

Explicit register/unregister

hooks = HighflameCrewHooks(client, mode="enforce")
hooks.register()

try:
    result = crew.kickoff()
finally:
    hooks.unregister()

Context manager (recommended)

with HighflameCrewHooks(client, mode="enforce") as hooks:
    result = crew.kickoff()

The context manager registers on entry and unregisters on exit, even if an exception is raised.

TypeScript

Enforcement Modes

Python

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

TypeScript

In monitor and alert mode, BlockedError is never raised. The crew proceeds normally regardless of policy decisions.

Complete Example

Python

from crewai import Agent, Task, Crew, Process
from highflame import Highflame, BlockedError
from highflame.integrations.crewai import HighflameCrewHooks

client = Highflame(api_key="hf_sk_...")

researcher = Agent(
    role="Research Analyst",
    goal="Research the given topic thoroughly",
    tools=[web_search_tool, file_read_tool],
)

research_task = Task(
    description="Research the latest developments in {topic}",
    agent=researcher,
)

crew = Crew(
    agents=[researcher],
    tasks=[research_task],
    process=Process.sequential,
)

try:
    with HighflameCrewHooks(client, mode="enforce") as hooks:
        result = crew.kickoff(inputs={"topic": user_topic})
except BlockedError as e:
    print(f"Crew execution blocked: {e.response.policy_reason}")

TypeScript

Session Tracking Across Crew Runs

When running multiple crews for the same user session, pass an explicit session_id to correlate risk signals:

Python

session_id = f"user_{user_id}_session_{conversation_id}"

with HighflameCrewHooks(client, session_id=session_id) as hooks:
    result = crew.kickoff()

TypeScript

Error Handling

Python

from highflame import BlockedError

try:
    with HighflameCrewHooks(client) as hooks:
        result = crew.kickoff()
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
crewai	1.0+
highflame	latest

TypeScript