agentconnect.utils package

Utility functions for the AgentConnect framework.

This module provides various utility functions and classes used throughout the framework, including interaction control for rate limiting, token usage tracking, and logging configuration.

Key components:

• InteractionControl: Controls agent interactions with rate limiting and turn tracking

• InteractionState: Enum for interaction states (CONTINUE, STOP, WAIT)

• TokenConfig: Configuration for token-based rate limiting

• Logging utilities: Configurable logging setup with colored output

• Wallet management: Functions for handling agent wallet configurations and data

class InteractionControl(agent_id, token_config, max_turns=20, current_turn=0, last_interaction_time=1746183145.892909, _cooldown_callback=None, _conversation_stats=None)

Bases: object

High-level control for agent interactions.

This class provides rate limiting, turn tracking, and conversation statistics for agent interactions.

Parameters:

• agent_id (str)

• token_config (TokenConfig)

• max_turns (int)

• current_turn (int)

• last_interaction_time (float)

• _cooldown_callback (Callable[[int], None] | None)

• _conversation_stats (Dict[str, Dict[str, Any]])

agent_id

The ID of the agent this control belongs to.

token_config

Configuration for token-based rate limiting

max_turns

Maximum number of turns in a conversation

current_turn

Current turn number

last_interaction_time

Timestamp of the last interaction

_cooldown_callback

Optional callback function to call when cooldown is triggered

_conversation_stats

Dictionary of conversation statistics

__post_init__()

Initialize conversation stats dictionary.

current_turn: int = 0

get_callback_handlers()

Create a list of callback handlers managed by InteractionControl (primarily rate limiting).

Return type:

List[BaseCallbackHandler]

Returns:

List containing configured BaseCallbackHandler instances.

get_conversation_stats(conversation_id=None)

Get statistics for a specific conversation or all conversations.

Parameters:

conversation_id (Optional[str]) – Optional ID of the conversation to get stats for

Return type:

Dict[str, Any]

Returns:

Dictionary of conversation statistics

last_interaction_time: float = 1746183145.892909

max_turns: int = 20

async process_interaction(token_count, conversation_id=None)

Process an interaction and return the state.

Parameters:

• token_count (int) – Number of tokens used in the interaction

• conversation_id (Optional[str]) – Optional conversation ID for tracking stats

Return type:

InteractionState

Returns:

InteractionState indicating whether to continue, stop, or wait

reset_turn_counter()

Reset the turn counter to zero.

Return type:

None

set_cooldown_callback(callback)

Set a callback function to be called when cooldown is triggered.

Parameters:

callback (Callable[[int], None]) – Function that takes cooldown duration in seconds as argument

Return type:

None

agent_id: str

token_config: TokenConfig

class InteractionState(value)

Bases: Enum

Enum for interaction states.

CONTINUE

Continue the interaction

STOP

Stop the interaction

WAIT

Wait before continuing the interaction

CONTINUE = 'continue'

STOP = 'stop'

WAIT = 'wait'

class TokenConfig(max_tokens_per_minute, max_tokens_per_hour, current_minute_tokens=0, current_hour_tokens=0, last_minute_reset=1746183145.8925169, last_hour_reset=1746183145.8925173)

Bases: object

Configuration for token-based rate limiting.

Parameters:

• max_tokens_per_minute (int)

• max_tokens_per_hour (int)

• current_minute_tokens (int)

• current_hour_tokens (int)

• last_minute_reset (float)

• last_hour_reset (float)

max_tokens_per_minute

Maximum tokens allowed per minute

max_tokens_per_hour

Maximum tokens allowed per hour

current_minute_tokens

Current token count for the minute

current_hour_tokens

Current token count for the hour

last_minute_reset

Timestamp of the last minute reset

last_hour_reset

Timestamp of the last hour reset

add_tokens(token_count)

Add tokens to the current count.

Parameters:

token_count (int) – Number of tokens to add

Return type:

None

current_hour_tokens: int = 0

current_minute_tokens: int = 0

get_cooldown_duration()

Get the cooldown duration in seconds if rate limits are exceeded.

Return type:

Optional[int]

Returns:

Cooldown duration in seconds, or None if no cooldown is needed

last_hour_reset: float = 1746183145.8925173

last_minute_reset: float = 1746183145.8925169

max_tokens_per_minute: int

max_tokens_per_hour: int

class RateLimitingCallbackHandler(max_tokens_per_minute=5500, max_tokens_per_hour=100000, cooldown_callback=None)

Bases: BaseCallbackHandler

Callback handler that implements rate limiting for LLM calls.

This handler tracks token usage and enforces rate limits for LLM calls, triggering cooldown periods when limits are reached.

Parameters:

• max_tokens_per_minute (int)

• max_tokens_per_hour (int)

• cooldown_callback (Callable[[int], None] | None)

max_tokens_per_minute

Maximum tokens allowed per minute

max_tokens_per_hour

Maximum tokens allowed per hour

current_minute_tokens

Current token count for the minute

current_hour_tokens

Current token count for the hour

last_minute_reset

Timestamp of the last minute reset

last_hour_reset

Timestamp of the last hour reset

in_cooldown

Whether the handler is in cooldown

cooldown_until

Timestamp when cooldown ends

cooldown_callback

Optional callback function to call when cooldown is triggered

on_chain_end(outputs, **kwargs)

Handle chain end event.

Parameters:

• outputs (Dict[str, Any]) – Chain outputs

• **kwargs (Any) – Additional arguments

Return type:

None

on_llm_end(response, **kwargs)

Handle LLM end event.

Parameters:

• response – LLM response

• **kwargs (Any) – Additional arguments

Return type:

None

on_llm_start(serialized, prompts, **kwargs)

Handle LLM start event.

Parameters:

• serialized (Dict[str, Any]) – Serialized LLM data

• prompts (List[str]) – List of prompts

• **kwargs (Any) – Additional arguments

Return type:

None

setup_logging(level=LogLevel.INFO, module_levels=None)

Configure logging with colors and per-module settings.

Parameters:

• level (LogLevel) – Default log level for all modules

• module_levels (Optional[Dict[str, LogLevel]]) – Dict of module names and their specific log levels

Return type:

None

class LogLevel(value)

Bases: Enum

Enum for log levels.

DEBUG

Debug log level

INFO

Info log level

WARNING

Warning log level

ERROR

Error log level

DEBUG = 10

INFO = 20

WARNING = 30

ERROR = 40

disable_all_logging()

Disable all logging output.

This is useful for examples and tests where logging output is not needed.

Return type:

None

get_module_levels_for_development()

Get recommended log levels for development.

Return type:

Dict[str, LogLevel]

Returns:

Dictionary of module names and their recommended log levels for development

load_wallet_data(agent_id, data_dir=None)

Loads previously persisted wallet data for an agent.

Parameters:

• agent_id (str) – String identifier for the agent.

• data_dir (Union[str, Path, None]) – Optional custom directory for wallet data storage. If None, uses the DEFAULT_DATA_DIR.

Return type:

Optional[str]

Returns:

The loaded wallet data as a JSON string if the file exists, otherwise None.

Raises:

IOError – If the file exists but can’t be read properly.

save_wallet_data(agent_id, wallet_data, data_dir=None)

Persists the exported wallet data for an agent, allowing the agent to retain access to the same wallet across restarts.

SECURITY NOTE: This default implementation stores wallet data unencrypted on disk, which is suitable for testing/demo but NOT secure for production environments holding real assets. Real-world applications should encrypt this data.

Parameters:

• agent_id (str) – String identifier for the agent.

• wallet_data (Union[WalletData, str, Dict]) – The wallet data to save. Can be a cdp.WalletData object, a Dict representation, or a JSON string.

• data_dir (Union[str, Path, None]) – Optional custom directory for wallet data storage. If None, uses the DEFAULT_DATA_DIR.

Raises:

IOError – If the data directory can’t be created or the file can’t be written.

Return type:

None

set_wallet_data_dir(data_dir)

Set a custom directory for wallet data storage.

Parameters:

data_dir (Union[str, Path]) – Path to the directory where wallet data will be stored Can be a string or Path object

Return type:

Path

Returns:

Path object pointing to the created directory

Raises:

IOError – If the directory can’t be created

set_default_data_dir(data_dir)

Set the default directory for wallet data storage globally.

Parameters:

data_dir (Union[str, Path]) – Path to the directory where wallet data will be stored Can be a string or Path object

Return type:

Path

Returns:

Path object pointing to the created directory

Raises:

IOError – If the directory can’t be created

wallet_exists(agent_id, data_dir=None)

Check if wallet data exists for a specific agent.

Parameters:

• agent_id (str) – String identifier for the agent.

• data_dir (Union[str, Path, None]) – Optional custom directory for wallet data storage. If None, uses the DEFAULT_DATA_DIR.

Return type:

bool

Returns:

True if wallet data exists, False otherwise.

delete_wallet_data(agent_id, data_dir=None)

Delete wallet data for a specific agent.

Parameters:

• agent_id (str) – String identifier for the agent.

• data_dir (Union[str, Path, None]) – Optional custom directory for wallet data storage. If None, uses the DEFAULT_DATA_DIR.

Return type:

bool

Returns:

True if wallet data was successfully deleted, False otherwise.

get_all_wallets(data_dir=None)

Get information about all wallet files in the specified directory.

Parameters:

data_dir (Union[str, Path, None]) – Optional custom directory for wallet data storage. If None, uses the DEFAULT_DATA_DIR.

Return type:

List[Dict]

Returns:

List of dictionaries with wallet information (agent_id, file_path, etc.)

class ToolTracerCallbackHandler(agent_id, print_tool_activity=True, print_reasoning_steps=True)

Bases: BaseCallbackHandler

Callback handler for tracing agent activity. Logs detailed activity and optionally prints concise updates (like LLM generation text, tool usage, etc.) to the console based on configuration.

Parameters:

• agent_id (str)

• print_tool_activity (bool)

• print_reasoning_steps (bool)

on_agent_action(action, *, run_id, parent_run_id=None, **kwargs)

Run on agent action.

Parameters:

• action (AgentAction) – The agent action.

• run_id (UUID) – The run ID. This is the ID of the current run.

• parent_run_id (UUID) – The parent run ID. This is the ID of the parent run.

• kwargs (Any) – Additional keyword arguments.

Return type:

Any

on_chain_end(outputs, **kwargs)

Print the agent’s ‘thought’ steps (reasoning before tool calls) in a clean, sequential manner. Handles different AIMessage structures from various models (e.g., Gemini, Anthropic). Uses REASONING_COLOR for output and msg.id for deduplication.

on_tool_end(output, *, run_id, parent_run_id=None, tags=None, name='UnknownTool', **kwargs)

Run when the tool ends running.

Parameters:

• output (Any) – The output of the tool.

• run_id (UUID) – The run ID. This is the ID of the current run.

• parent_run_id (UUID) – The parent run ID. This is the ID of the parent run.

• kwargs (Any) – Additional keyword arguments.

• tags (List[str] | None)

• name (str)

Return type:

Any

on_tool_error(error, *, run_id, parent_run_id=None, tags=None, name='UnknownTool', **kwargs)

Run when tool errors.

Parameters:

• error (BaseException) – The error that occurred.

• run_id (UUID) – The run ID. This is the ID of the current run.

• parent_run_id (UUID) – The parent run ID. This is the ID of the parent run.

• kwargs (Any) – Additional keyword arguments.

• tags (List[str] | None)

• name (str)

Return type:

Any

on_tool_start(serialized, input_str, *, run_id, parent_run_id=None, tags=None, metadata=None, inputs=None, **kwargs)

Run when the tool starts running.

Parameters:

• serialized (Dict[str, Any]) – The serialized tool.

• input_str (str) – The input string.

• run_id (UUID) – The run ID. This is the ID of the current run.

• parent_run_id (UUID) – The parent run ID. This is the ID of the parent run.

• tags (Optional[List[str]]) – The tags.

• metadata (Optional[Dict[str, Any]]) – The metadata.

• inputs (Optional[Dict[str, Any]]) – The inputs.

• kwargs (Any) – Additional keyword arguments.

Return type:

Any

Submodules

• agentconnect.utils.callbacks module
  • ToolTracerCallbackHandler
    • ToolTracerCallbackHandler.on_tool_start()
    • ToolTracerCallbackHandler.on_tool_end()
    • ToolTracerCallbackHandler.on_tool_error()
    • ToolTracerCallbackHandler.on_agent_action()
    • ToolTracerCallbackHandler.on_chain_end()
• agentconnect.utils.interaction_control module
  • InteractionState
    • InteractionState.CONTINUE
    • InteractionState.STOP
    • InteractionState.WAIT
    • InteractionState.CONTINUE
    • InteractionState.STOP
    • InteractionState.WAIT
  • RateLimitingCallbackHandler
    • RateLimitingCallbackHandler.max_tokens_per_minute
    • RateLimitingCallbackHandler.max_tokens_per_hour
    • RateLimitingCallbackHandler.current_minute_tokens
    • RateLimitingCallbackHandler.current_hour_tokens
    • RateLimitingCallbackHandler.last_minute_reset
    • RateLimitingCallbackHandler.last_hour_reset
    • RateLimitingCallbackHandler.in_cooldown
    • RateLimitingCallbackHandler.cooldown_until
    • RateLimitingCallbackHandler.cooldown_callback
    • RateLimitingCallbackHandler.on_llm_start()
    • RateLimitingCallbackHandler.on_llm_end()
    • RateLimitingCallbackHandler.on_chain_end()
  • TokenConfig
    • TokenConfig.max_tokens_per_minute
    • TokenConfig.max_tokens_per_hour
    • TokenConfig.current_minute_tokens
    • TokenConfig.current_hour_tokens
    • TokenConfig.last_minute_reset
    • TokenConfig.last_hour_reset
    • TokenConfig.max_tokens_per_minute
    • TokenConfig.max_tokens_per_hour
    • TokenConfig.current_minute_tokens
    • TokenConfig.current_hour_tokens
    • TokenConfig.last_minute_reset
    • TokenConfig.last_hour_reset
    • TokenConfig.add_tokens()
    • TokenConfig.get_cooldown_duration()
  • InteractionControl
    • InteractionControl.agent_id
    • InteractionControl.token_config
    • InteractionControl.max_turns
    • InteractionControl.current_turn
    • InteractionControl.last_interaction_time
    • InteractionControl._cooldown_callback
    • InteractionControl._conversation_stats
    • InteractionControl.agent_id
    • InteractionControl.token_config
    • InteractionControl.max_turns
    • InteractionControl.current_turn
    • InteractionControl.last_interaction_time
    • InteractionControl.__post_init__()
    • InteractionControl.process_interaction()
    • InteractionControl.set_cooldown_callback()
    • InteractionControl.reset_turn_counter()
    • InteractionControl.get_conversation_stats()
    • InteractionControl.get_callback_handlers()
• agentconnect.utils.logging_config module
  • LogLevel
    • LogLevel.DEBUG
    • LogLevel.INFO
    • LogLevel.WARNING
    • LogLevel.ERROR
    • LogLevel.DEBUG
    • LogLevel.INFO
    • LogLevel.WARNING
    • LogLevel.ERROR
  • ColoredFormatter
    • ColoredFormatter.COLORS
    • ColoredFormatter.COLORS
    • ColoredFormatter.format()
  • setup_logging()
  • disable_all_logging()
  • get_module_levels_for_development()
  • setup_langgraph_logging()
• agentconnect.utils.payment_helper module
  • verify_payment_environment()
  • validate_cdp_environment()
  • get_wallet_metadata()
  • backup_wallet_data()
  • check_agent_payment_readiness()
• agentconnect.utils.wallet_manager module
  • set_default_data_dir()
  • set_wallet_data_dir()
  • save_wallet_data()
  • load_wallet_data()
  • wallet_exists()
  • get_all_wallets()
  • delete_wallet_data()