Hivetrace

https://pypi.org/project/hivetrace/
Admin
The Hivetrace SDK lets you integrate with the Hivetrace service to monitor user prompts and LLM...
Tokens:8,534
Snippets:96
Trust Score:10
Update:3 months ago
Show doc for...
Tokens:
Raw
### Install Hivetrace SDK

Source: https://pypi.org/project/hivetrace/1.4.1

Install the base version of the Hivetrace SDK using pip.

```bash
pip install hivetrace[base]
```

--------------------------------

### Install HiveTrace OpenAI Agents package

Source: https://pypi.org/project/hivetrace/1.4.1

Install the required package via pip.

```bash
pip install hivetrace[openai_agents]==1.3.5
```

--------------------------------

### Configure via .env File

Source: https://pypi.org/project/hivetrace/1.4.1

Example format for environment variables in a .env file.

```text
HIVETRACE_URL=https://your-hivetrace-instance.com
HIVETRACE_ACCESS_TOKEN=your-access-token  # obtained in the UI (API Tokens page)
```

--------------------------------

### Install HiveTrace LangChain dependencies

Source: https://pypi.org/project/hivetrace/1.4.1

Install the required HiveTrace package with LangChain support.

```bash
pip install hivetrace[langchain]>=1.3.5
# optional: add to requirements.txt and install
echo "hivetrace[langchain]>=1.3.3" >> requirements.txt
pip install -r requirements.txt
```

--------------------------------

### Install HiveTrace SDK

Source: https://pypi.org/project/hivetrace/1.4.1

Add the HiveTrace SDK with CrewAI support via pip or requirements.txt.

```bash
# Via pip (for quick testing)
pip install hivetrace[crewai]>=1.3.5

# Or add to requirements.txt (recommended)
echo "hivetrace[crewai]>=1.3.3" >> requirements.txt
pip install -r requirements.txt

```

--------------------------------

### Send User Prompt with Additional Parameters

Source: https://pypi.org/project/hivetrace/1.4.1

Example of calling the input method with session, user, and agent context, including the censor_only flag.

```python
response = client.input(
    application_id="your-application-id",
    message="User prompt here",
    additional_parameters={
        "session_id": "your-session-id",
        "user_id": "your-user-id",
        "agents": {
            "agent-1-id": {"name": "Agent 1", "description": "Agent description"},
            "agent-2-id": {"name": "Agent 2"},
            "agent-3-id": {}
        },
        # If you want to send only to censor and avoid DB persistence on backend
        "censor_only": True,
    }
)
```

--------------------------------

### Hivetrace Input Response Example

Source: https://pypi.org/project/hivetrace/1.4.1

JSON structure returned after processing a user prompt.

```json
{
  "blocked": false,
  "status": "processed",
  "monitoring_result": {
    "is_toxic": false,
    "type_of_violation": "benign",
    "token_count": 9,
    "token_usage_severity": None
  }
}
```

--------------------------------

### Hivetrace Output Response Example

Source: https://pypi.org/project/hivetrace/1.4.1

JSON structure returned after processing an LLM response.

```json
{
  "blocked": false,
  "status": "processed",
  "monitoring_result": {
    "is_toxic": false,
    "type_of_violation": "safe",
    "token_count": 21,
    "token_usage_severity": None
  }
}
```

--------------------------------

### Configure Client Explicitly

Source: https://pypi.org/project/hivetrace/1.4.1

Passes a configuration dictionary directly to the client constructor.

```python
client = SyncHivetraceSDK(
    config={
        "HIVETRACE_URL": HIVETRACE_URL,
        "HIVETRACE_ACCESS_TOKEN": HIVETRACE_ACCESS_TOKEN,
    },
)
```

--------------------------------

### Attach Trace Processor

Source: https://pypi.org/project/hivetrace/1.4.1

Register the trace processor at application startup. Use the explicit configuration if environment variables are not preferred.

```python
from agents import set_trace_processors
from hivetrace.adapters.openai_agents.tracing import HivetraceOpenAIAgentProcessor

set_trace_processors([
    HivetraceOpenAIAgentProcessor()  # will take config from env
])
```

```python
from agents import set_trace_processors
from hivetrace import SyncHivetraceSDK
from hivetrace.adapters.openai_agents.tracing import HivetraceOpenAIAgentProcessor

hivetrace = SyncHivetraceSDK(config={
    "HIVETRACE_URL": "http://localhost:8000",
    "HIVETRACE_ACCESS_TOKEN": "ht_...",
})

set_trace_processors([
    HivetraceOpenAIAgentProcessor(
        application_id="00000000-0000-0000-0000-000000000000",
        hivetrace_instance=hivetrace,
    )
])
```

--------------------------------

### Import Hivetrace Clients

Source: https://pypi.org/project/hivetrace/1.4.1

Import the synchronous and asynchronous client classes to begin monitoring LLM workflows.

```python
from hivetrace import SyncHivetraceSDK, AsyncHivetraceSDK
```

--------------------------------

### Send Asynchronous Requests

Source: https://pypi.org/project/hivetrace/1.4.1

Demonstrates using the AsyncHivetraceSDK with async/await syntax.

```python
import asyncio

async def main():
    # option 1: context manager
    async with AsyncHivetraceSDK() as client:
        response = await client.input(
            application_id="your-application-id",
            message="User prompt here",
        )

    # option 2: manual close
    client = AsyncHivetraceSDK()
    try:
        response = await client.input(
            application_id="your-application-id",
            message="User prompt here",
        )
    finally:
        await client.close()

asyncio.run(main())
```

--------------------------------

### Initialize Synchronous Hivetrace Client

Source: https://pypi.org/project/hivetrace/1.4.1

Initializes the synchronous client, which reads configuration from environment variables or explicit settings.

```python
# The sync client reads configuration from environment variables or accepts an explicit config
client = SyncHivetraceSDK()
```

--------------------------------

### Initialize Asynchronous Hivetrace Client

Source: https://pypi.org/project/hivetrace/1.4.1

Initializes the asynchronous client, which can be used as a context manager.

```python
# The async client can be used as a context manager
client = AsyncHivetraceSDK()
```

--------------------------------

### Initialize HiveTrace in FastAPI

Source: https://pypi.org/project/hivetrace/1.4.1

Use the FastAPI lifespan context manager to initialize the HiveTrace SDK and ensure the connection is closed on shutdown.

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
# ADD: import HiveTrace SDK
from hivetrace import SyncHivetraceSDK
from src.config import HIVETRACE_ACCESS_TOKEN, HIVETRACE_URL

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ADD: initialize HiveTrace client
    hivetrace = SyncHivetraceSDK(
        config={
            "HIVETRACE_URL": HIVETRACE_URL,
            "HIVETRACE_ACCESS_TOKEN": HIVETRACE_ACCESS_TOKEN,
        }
    )
    # Store client in app state
    app.state.hivetrace = hivetrace
    try:
        yield
    finally:
        # IMPORTANT: close connection on shutdown
        hivetrace.close()

app = FastAPI(lifespan=lifespan)
```

--------------------------------

### Integrate HiveTrace with LangChain

Source: https://pypi.org/project/hivetrace/1.4.1

Demonstrates using run_with_tracing to execute an orchestrator agent with automated logging.

```python
import os, uuid
from langchain_openai import ChatOpenAI
from src.core.orchestrator import OrchestratorAgent
from src.core.constants import PREDEFINED_AGENT_IDS
from hivetrace.adapters.langchain import run_with_tracing

llm = ChatOpenAI(model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"), temperature=0.2, streaming=False)
orchestrator = OrchestratorAgent(llm, predefined_agent_ids=PREDEFINED_AGENT_IDS)

result = run_with_tracing(
    orchestrator=orchestrator,
    query="Format this text and count the number of words",
    application_id=os.getenv("HIVETRACE_APP_ID"),
    user_id=os.getenv("USER_ID"),
    session_id=os.getenv("SESSION_ID"),
    conversation_id=str(uuid.uuid4()),
)
print(result)
```

```python
from fastapi import APIRouter, Request
from hivetrace.adapters.langchain import run_with_tracing
import uuid

router = APIRouter()

@router.post("/query")
async def process_query(payload: dict, request: Request):
    orchestrator = request.app.state.orchestrator
    conv_id = str(uuid.uuid4()) # always create a new agent_conversation_id for each request to group agent work for the same question
    result = run_with_tracing(
        orchestrator=orchestrator,
        query=payload["query"],
        application_id=request.app.state.HIVETRACE_APP_ID,
        user_id=request.app.state.USER_ID,
        session_id=request.app.state.SESSION_ID,
        conversation_id=conv_id,
    )
    return {"status": "success", "result": result}
```

--------------------------------

### Send Synchronous Requests

Source: https://pypi.org/project/hivetrace/1.4.1

Demonstrates using the SyncHivetraceSDK via a context manager or manual lifecycle management.

```python
def main():
    # option 1: context manager
    with SyncHivetraceSDK() as client:
        response = client.input(
            application_id="your-application-id",
            message="User prompt here",
        )

    # option 2: manual close
    client = SyncHivetraceSDK()
    try:
        response = client.input(
            application_id="your-application-id",
            message="User prompt here",
        )
    finally:
        client.close()

main()
```

--------------------------------

### input(application_id, message, additional_parameters, files)

Source: https://pypi.org/project/hivetrace/1.4.1

Sends a user prompt to Hivetrace for monitoring. Supports optional session context, agent definitions, and file attachments.

```APIDOC
## input(application_id, message, additional_parameters, files)

### Description
Sends a user prompt to Hivetrace. This method is available in both synchronous and asynchronous versions.

### Parameters
- **application_id** (str) - Required - Application identifier (must be a valid UUID).
- **message** (str) - Required - The user prompt.
- **additional_parameters** (dict) - Optional - Extra context including session_id, user_id, agents, and special flags like 'censor_only'.
- **files** (list[tuple[str, bytes, str]]) - Optional - List of tuples containing (filename, content, mime_type).

### Response
- **blocked** (bool) - Indicates role restrictions.
- **status** (str) - Processing status.
- **monitoring_result** (dict) - Contains toxicity and violation details.

### Response Example
{
  "blocked": false,
  "status": "processed",
  "monitoring_result": {
    "is_toxic": false,
    "type_of_violation": "benign",
    "token_count": 9,
    "token_usage_severity": null
  }
}
```

--------------------------------

### Configure HiveTrace Environment Variables

Source: https://pypi.org/project/hivetrace/1.4.1

Define connection details and credentials in a .env file.

```text
# .env file
HIVETRACE_URL=https://your-hivetrace-instance.com
HIVETRACE_ACCESS_TOKEN=your-access-token
HIVETRACE_APP_ID=your-application-id

```

--------------------------------

### Configure environment variables

Source: https://pypi.org/project/hivetrace/1.4.1

Set the necessary environment variables for HiveTrace and OpenAI connectivity.

```bash
HIVETRACE_URL=http://localhost:8000          # Your HiveTrace URL
HIVETRACE_ACCESS_TOKEN=ht_...                # Your HiveTrace access token
HIVETRACE_APPLICATION_ID=00000000-...-0000   # Your HiveTrace application ID

SESSION_ID=
USERID=

OPENAI_API_KEY=
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini
```

--------------------------------

### Send Input with Synchronous Client

Source: https://pypi.org/project/hivetrace/1.4.1

Sends a user prompt to the specified application ID, with optional support for file attachments.

```python
response = client.input(
    application_id="your-application-id",  # Obtained after registering the application in the UI
    message="User prompt here",
)

# Optionally attach files (filename, bytes, mime_type)
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = client.input(
    application_id="your-application-id",
    message="User prompt with files",
    files=files,
)
```

--------------------------------

### AsyncHivetraceSDK.input

Source: https://pypi.org/project/hivetrace/1.4.1

Sends a user prompt to the Hivetrace application using the asynchronous client.

```APIDOC
## AsyncHivetraceSDK.input

### Description
Asynchronously sends a user prompt to the specified application. Supports optional file attachments.

### Parameters
- **application_id** (str) - Required - The ID of the application.
- **message** (str) - Required - The user prompt message.
- **files** (list) - Optional - A list of tuples containing (filename, file_object, mime_type).
```

--------------------------------

### SyncHivetraceSDK.input

Source: https://pypi.org/project/hivetrace/1.4.1

Sends a user prompt to the Hivetrace application using the synchronous client.

```APIDOC
## SyncHivetraceSDK.input

### Description
Sends a user prompt to the specified application. Supports optional file attachments.

### Parameters
- **application_id** (str) - Required - The ID of the application.
- **message** (str) - Required - The user prompt message.
- **files** (list) - Optional - A list of tuples containing (filename, file_object, mime_type).
```

--------------------------------

### Reuse HiveTrace Client in FastAPI

Source: https://pypi.org/project/hivetrace/1.4.1

Shows how to manage the lifecycle of a SyncHivetraceSDK client using FastAPI lifespan events and passing it to tracing helpers.

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from hivetrace import SyncHivetraceSDK

@asynccontextmanager
async def lifespan(app: FastAPI):
    hivetrace = SyncHivetraceSDK()
    app.state.hivetrace = hivetrace
    try:
        yield
    finally:
        hivetrace.close()

app = FastAPI(lifespan=lifespan)
```

```python
result = run_with_tracing(
    orchestrator=orchestrator,
    query=payload.query,
    hivetrace=request.app.state.hivetrace,  # pass your own client
    application_id=request.app.state.HIVETRACE_APP_ID,
)
```

--------------------------------

### Attach AgentLoggingCallback to LangChain executor

Source: https://pypi.org/project/hivetrace/1.4.1

Initialize the logging callback and pass it to the AgentExecutor to track agent interactions.

```python
from hivetrace.adapters.langchain import AgentLoggingCallback
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

class OrchestratorAgent:
    def __init__(self, llm, predefined_agent_ids=None):
        self.llm = llm
        self.logging_callback = AgentLoggingCallback(
            default_root_name="MainHub",
            predefined_agent_ids=predefined_agent_ids,
        )
        # Example: wrapper agents as tools
        # MathAgentTool/TextAgentTool internally pass self.logging_callback further
        agent = create_openai_tools_agent(self.llm, self.tools, ChatPromptTemplate.from_messages([
            ("system", "You are the orchestrator agent of a multi-agent system."),
            MessagesPlaceholder(variable_name="chat_history", optional=True),
            ("human", "{input}"),
            MessagesPlaceholder(variable_name="agent_scratchpad"),
        ]))
        self.executor = AgentExecutor(
            agent=agent,
            tools=self.tools,
            verbose=True,
            callbacks=[self.logging_callback],
        )
```

--------------------------------

### output(application_id, message, additional_parameters, files)

Source: https://pypi.org/project/hivetrace/1.4.1

Sends an LLM response to Hivetrace for monitoring. Supports optional session context and file attachments.

```APIDOC
## output(application_id, message, additional_parameters, files)

### Description
Sends an LLM response to Hivetrace. This method is available in both synchronous and asynchronous versions.

### Parameters
- **application_id** (str) - Required - Application identifier (must be a valid UUID).
- **message** (str) - Required - The LLM response.
- **additional_parameters** (dict) - Optional - Extra context including session, user, and agents.
- **files** (list[tuple[str, bytes, str]]) - Optional - List of tuples containing (filename, content, mime_type).

### Response
- **blocked** (bool) - Indicates role restrictions.
- **status** (str) - Processing status.
- **monitoring_result** (dict) - Contains toxicity and violation details.

### Response Example
{
  "blocked": false,
  "status": "processed",
  "monitoring_result": {
    "is_toxic": false,
    "type_of_violation": "safe",
    "token_count": 21,
    "token_usage_severity": null
  }
}
```

--------------------------------

### Send Input with Asynchronous Client

Source: https://pypi.org/project/hivetrace/1.4.1

Sends a user prompt asynchronously to the specified application ID, with optional support for file attachments.

```python
response = await client.input(
    application_id="your-application-id",
    message="User prompt here",
)

# With files (filename, bytes, mime_type)
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = await client.input(
    application_id="your-application-id",
    message="User prompt with files",
    files=files,
)
```

--------------------------------

### Send Output with Synchronous Client

Source: https://pypi.org/project/hivetrace/1.4.1

Sends an LLM response to the specified application ID, with optional support for file attachments.

```python
response = client.output(
    application_id="your-application-id",
    message="LLM response here",
)

# With files
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = client.output(
    application_id="your-application-id",
    message="LLM response with files",
    files=files,
)
```

--------------------------------

### Map Agent Roles to HiveTrace IDs

Source: https://pypi.org/project/hivetrace/1.4.1

Create a mapping dictionary to link CrewAI agent roles to their respective HiveTrace IDs.

```python
from crewai import Agent
# ADD: import agent IDs
from src.config import EDITOR_ID, PLANNER_ID, WRITER_ID

# ADD: mapping for HiveTrace (REQUIRED!)
agent_id_mapping = {
    "Content Planner": {  # ← Exactly the same as Agent(role="Content Planner")
        "id": PLANNER_ID,
        "description": "Creates content plans"
    },
    "Content Writer": {   # ← Exactly the same as Agent(role="Content Writer")
        "id": WRITER_ID,
        "description": "Writes high-quality articles"
    },
    "Editor": {           # ← Exactly the same as Agent(role="Editor")
        "id": EDITOR_ID,
        "description": "Edits and improves articles"
    },
}

# Your existing agents (NO CHANGES)
planner = Agent(
    role="Content Planner",  # ← Must match key in agent_id_mapping
    goal="Create a structured content plan for the given topic",
    backstory="You are an experienced analyst...",
    verbose=True,
)

writer = Agent(
    role="Content Writer",   # ← Must match key in agent_id_mapping
    goal="Write an informative and engaging article",
    backstory="You are a talented writer...",
    verbose=True,
)

editor = Agent(
    role="Editor",           # ← Must match key in agent_id_mapping
    goal="Improve the article",
    backstory="You are an experienced editor...",
    verbose=True,
)

```

--------------------------------

### Send Output with Asynchronous Client

Source: https://pypi.org/project/hivetrace/1.4.1

Sends an LLM response asynchronously to the specified application ID, with optional support for file attachments.

```python
response = await client.output(
    application_id="your-application-id",
    message="LLM response here",
)

# With files
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = await client.output(
    application_id="your-application-id",
    message="LLM response with files",
    files=files,
)
```

--------------------------------

### AsyncHivetraceSDK.output

Source: https://pypi.org/project/hivetrace/1.4.1

Sends an LLM response to the Hivetrace application using the asynchronous client.

```APIDOC
## AsyncHivetraceSDK.output

### Description
Asynchronously sends an LLM response to the specified application. Supports optional file attachments.

### Parameters
- **application_id** (str) - Required - The ID of the application.
- **message** (str) - Required - The LLM response message.
- **files** (list) - Optional - A list of tuples containing (filename, file_object, mime_type).
```

--------------------------------

### Hivetrace Input Method Definitions

Source: https://pypi.org/project/hivetrace/1.4.1

Synchronous and asynchronous method signatures for the input function.

```python
# Sync
def input(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> dict: ...

# Async
async def input(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> dict: ...
```

--------------------------------

### Monitor Agent Execution

Source: https://pypi.org/project/hivetrace/1.4.1

Comparison of agent execution before and after adding HiveTrace monitoring.

```python
from agents import Agent, Runner

assistant = Agent(name="Assistant", instructions="Be helpful.")
print(Runner.run_sync(assistant, "Hi!"))
```

```python
from agents import Agent, Runner, set_trace_processors
from hivetrace.adapters.openai_agents.tracing import HivetraceOpenAIAgentProcessor

set_trace_processors([HivetraceOpenAIAgentProcessor()])

assistant = Agent(name="Assistant", instructions="Be helpful.")
print(Runner.run_sync(assistant, "Hi!"))
```

--------------------------------

### Assign agent IDs to tools

Source: https://pypi.org/project/hivetrace/1.4.1

Assign the appropriate agent ID to tool instances before passing them to the Agent configuration.

```python
from src.tools import WordCountTool
from src.config import PLANNER_ID, WRITER_ID, EDITOR_ID

# ADD: create tools for each agent
planner_tools = [WordCountTool()]
writer_tools = [WordCountTool()]
editor_tools = [WordCountTool()]

# ADD: assign tools to agents
for tool in planner_tools:
    tool.agent_id = PLANNER_ID

for tool in writer_tools:
    tool.agent_id = WRITER_ID

for tool in editor_tools:
    tool.agent_id = EDITOR_ID

# Use tools in agents
planner = Agent(
    role="Content Planner",
    tools=planner_tools,  # ← Agent-specific tools
    # ... other parameters
)
```

--------------------------------

### Integrate HiveTrace with CrewAI

Source: https://pypi.org/project/hivetrace/1.4.1

Wrap the Crew creation function with the @trace decorator and log user requests using the HiveTrace SDK. Ensure the @trace decorator is applied to the function that creates and returns the Crew object.

```python
import uuid
from typing import Optional
from crewai import Crew
# ADD: HiveTrace imports
from hivetrace import SyncHivetraceSDK
from hivetrace import crewai_trace as trace

from src.agents import agent_id_mapping, planner, writer, editor
from src.tasks import plan_task, write_task, edit_task
from src.config import HIVETRACE_APP_ID

def process_topic(
    topic: str,
    hivetrace: SyncHivetraceSDK,  # ← ADD parameter
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
):
    # ADD: generate unique conversation ID
    agent_conversation_id = str(uuid.uuid4())
    
    # ADD: common trace parameters
    common_params = {
        "agent_conversation_id": agent_conversation_id,
        "user_id": user_id,
        "session_id": session_id,
    }

    # ADD: log user request
    hivetrace.input(
        application_id=HIVETRACE_APP_ID,
        message=f"Requesting information from agents on topic: {topic}",
        additional_parameters={
            **common_params,
            "agents": agent_id_mapping,  # ← pass agent mapping
        },
    )

    # ADD: @trace decorator for monitoring Crew
    @trace(
        hivetrace=hivetrace,
        application_id=HIVETRACE_APP_ID,
        agent_id_mapping=agent_id_mapping,  # ← REQUIRED!
    )
    def create_crew():
        return Crew(
            agents=[planner, writer, editor],
            tasks=[plan_task, write_task, edit_task],
            verbose=True,
        )

    # Execute with monitoring
    crew = create_crew()
    result = crew.kickoff(
        inputs={"topic": topic},
        **common_params  # ← pass common parameters
    )

    return {
        "result": result.raw,
        "execution_details": {**common_params, "status": "completed"},
    }
```

--------------------------------

### SyncHivetraceSDK.output

Source: https://pypi.org/project/hivetrace/1.4.1

Sends an LLM response to the Hivetrace application using the synchronous client.

```APIDOC
## SyncHivetraceSDK.output

### Description
Sends an LLM response to the specified application. Supports optional file attachments.

### Parameters
- **application_id** (str) - Required - The ID of the application.
- **message** (str) - Required - The LLM response message.
- **files** (list) - Optional - A list of tuples containing (filename, file_object, mime_type).
```

--------------------------------

### Trace Tool Invocations

Source: https://pypi.org/project/hivetrace/1.4.1

Decorate functions with @function_tool to enable automatic tracing of tool calls.

```python
from agents import function_tool

@function_tool(description_override="Adds two numbers")
def calculate_sum(a: int, b: int) -> int:
    return a + b
```

--------------------------------

### Hivetrace Output Method Definitions

Source: https://pypi.org/project/hivetrace/1.4.1

Synchronous and asynchronous method signatures for the output function.

```python
# Sync
def output(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> dict: ...

# Async
async def output(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> dict: ...
```

--------------------------------

### Define Unique Agent IDs

Source: https://pypi.org/project/hivetrace/1.4.1

Assign unique UUIDs to each agent in your configuration file.

```python
PLANNER_ID = "333e4567-e89b-12d3-a456-426614174001"
WRITER_ID = "444e4567-e89b-12d3-a456-426614174002"
EDITOR_ID = "555e4567-e89b-12d3-a456-426614174003"

```

--------------------------------

### Add HiveTrace support to custom tools

Source: https://pypi.org/project/hivetrace/1.4.1

Include an agent_id field in your BaseTool subclass to enable tracking of tool usage by specific agents.

```python
from crewai.tools import BaseTool
from typing import Optional

class WordCountTool(BaseTool):
    name: str = "WordCountTool"
    description: str = "Count words, characters and sentences in text"
    # ADD: HiveTrace field (REQUIRED!)
    agent_id: Optional[str] = None
    
    def _run(self, text: str) -> str:
        word_count = len(text.split())
        return f"Word count: {word_count}"
```

--------------------------------

### Pass HiveTrace client to FastAPI endpoint

Source: https://pypi.org/project/hivetrace/1.4.1

Inject the HiveTrace client from the application state into business logic to enable monitoring.

```python
from fastapi import APIRouter, Body, Request
# ADD: import HiveTrace type
from hivetrace import SyncHivetraceSDK

from src.services.topic_service import process_topic
from src.config import SESSION_ID, USER_ID

router = APIRouter(prefix="/api")

@router.post("/process-topic")
async def api_process_topic(request: Request, request_body: dict = Body(...)):
    # ADD: get HiveTrace client from app state
    hivetrace: SyncHivetraceSDK = request.app.state.hivetrace
    
    return process_topic(
        topic=request_body["topic"],
        hivetrace=hivetrace,  # ← pass client
        user_id=USER_ID,
        session_id=SESSION_ID,
    )
```

--------------------------------

### Define fixed agent UUIDs

Source: https://pypi.org/project/hivetrace/1.4.1

Map agent names to fixed UUIDs to ensure consistent identification in trace logs.

```python
PREDEFINED_AGENT_IDS = {
    "MainHub": "111e1111-e89b-12d3-a456-426614174099",
    "text_agent": "222e2222-e89b-12d3-a456-426614174099",
    "math_agent": "333e3333-e89b-12d3-a456-426614174099",
    "pre_text_agent": "444e4444-e89b-12d3-a456-426614174099",
    "pre_math_agent": "555e5555-e89b-12d3-a456-426614174099",
}
```

--------------------------------

### Close Async Client

Source: https://pypi.org/project/hivetrace/1.4.1

Explicitly closes an asynchronous client instance.

```python
await client.close()
```
Tokens:
Raw