anthropic.Utils

View module source on GitHub

​

AnthropicUtils

class AnthropicUtils()

Utility class for Anthropic API integration with Maxim.

This class provides static utility methods for parsing and processing Anthropic API requests and responses to integrate with Maxim’s logging and monitoring system. It handles message format conversion, parameter extraction, and response standardization.

All methods are static and can be called directly on the class without instantiation.

​

parse_message_param

@staticmethod
def parse_message_param(
        message: Iterable[MessageParam],
        override_role: Optional[str] = None) -> List[GenerationRequestMessage]

Parse Anthropic message parameters into Maxim format.

This method converts Anthropic MessageParam objects into Maxim’s GenerationRequestMessage format for consistent logging and tracking. It handles various message formats including string content and structured content blocks.

Arguments:

to be parsed. Can contain strings, dicts, or MessageParam objects.

• override_role Optional[str] - Optional role to override the message role. If provided, all messages will use this role instead of their original role.

Returns:

with role and content extracted and standardized.

Notes:

• String messages are treated as user messages by default
• Dict messages should have ‘role’ and ‘content’ keys
• Content blocks are flattened into text content
• Complex content structures are converted to string representation

​

get_model_params

@staticmethod
def get_model_params(max_tokens: int, **kwargs: Any) -> Dict[str, Any]

Extract and normalize model parameters for Maxim logging.

This method extracts relevant model parameters from Anthropic API calls and formats them for consistent logging in Maxim. It handles common parameters like temperature, top_p, and other generation settings.

Arguments:

model parameters such as temperature, top_p, top_k, system, etc.

Returns:

Dict[str, Any]: Dictionary containing normalized model parameters with non-None values only. Common parameters are extracted explicitly while additional parameters are included as-is.

Notes:

• Only non-None parameters are included in the result
• System, metadata, temperature, top_p, and top_k are handled explicitly
• Additional parameters from kwargs are included if they have values

​

parse_message_stream

@staticmethod
def parse_message_stream(stream: List[MessageStreamEvent]) -> Dict[str, Any]

Parse a list of Anthropic stream events into standardized format.

This method processes a complete stream of MessageStreamEvent objects and converts them into a standardized response format compatible with OpenAI-style responses for consistent logging and processing.

Arguments:

Anthropic’s streaming API response.

Returns:

Dict[str, Any]: Standardized response dictionary with the following structure:

• id: Unique identifier for the response
• created: Unix timestamp of creation
• choices: List with single choice containing message and finish_reason
• usage: Token usage statistics (prompt, completion, total)

Raises:

• ValueError - If the stream list is empty.

Notes:

• Text content is extracted from content_block_delta events
• Token usage is extracted from message_start events
• The response format mimics OpenAI’s API for compatibility
• Finish reason is set to “stop” by default (Anthropic doesn’t provide this directly)

​

parse_message

@staticmethod
def parse_message(message: Any) -> Dict[str, Any]

Parse an Anthropic Message response into standardized format.

This method converts an Anthropic Message object into a standardized response format compatible with OpenAI-style responses for consistent logging and processing across different AI providers.

Arguments:

with content, usage statistics, and metadata.

Returns:

Dict[str, Any]: Standardized response dictionary with the following structure:

• id: Message ID from Anthropic
• created: Unix timestamp of parsing time
• choices: List with single choice containing message and finish_reason
• usage: Token usage statistics (input, output, total tokens)

Notes:

• Content blocks are flattened into a single text string
• Both structured content blocks and dict-based content are supported
• Token usage is extracted from the message’s usage attribute
• Stop reason is mapped from Anthropic’s stop_reason or defaults to “stop”
• The response format mimics OpenAI’s API for cross-provider compatibility