Message

View module source on GitHub

​

MaximAnthropicMessages

class MaximAnthropicMessages(Messages)

Maxim-enhanced Anthropic Messages client.

This class extends the Anthropic Messages resource to integrate with Maxim’s logging and monitoring capabilities. It automatically tracks message creation, both streaming and non-streaming, and logs them through the Maxim platform.

The class handles trace management, generation logging, and error handling while maintaining compatibility with the original Anthropic Messages API.

Attributes:

​

__init__

def __init__(client: Anthropic, logger: Logger)

Initialize the Maxim Anthropic Messages client.

Arguments:

logging message interactions.

​

create_non_stream

def create_non_stream(*args, **kwargs) -> Any

Create a non-streaming message with Maxim logging.

This method handles non-streaming message creation while automatically logging the interaction through Maxim. It manages trace creation, generation tracking, and error handling.

Arguments:

Special headers:

• x-maxim-trace-id: Optional trace ID for associating with existing trace.
• x-maxim-generation-name: Optional name for the generation.

Returns:

Notes:

If logging fails, the method will still return the API response but will log a warning message.

​

create_stream

def create_stream(*args, **kwargs) -> Any

Create a streaming message with Maxim logging.

This method handles streaming message creation while automatically logging the interaction through Maxim. It manages trace creation, generation tracking, and processes streaming events.

Arguments:

Special headers:

• x-maxim-trace-id: Optional trace ID for associating with existing trace.
• x-maxim-generation-name: Optional name for the generation.

Returns:

handles logging of streaming events.

Notes:

The method returns a StreamWrapper that automatically processes stream chunks and logs the final result when the stream ends.

​

create

def create(*args,
           max_tokens: int,
           messages: Iterable[MessageParam],
           model: str,
           metadata: MetadataParam | NotGiven = NOT_GIVEN,
           stop_sequences: List[str] | NotGiven = NOT_GIVEN,
           system: Union[str, Iterable[TextBlockParam]] | NotGiven = NOT_GIVEN,
           temperature: float | NotGiven = NOT_GIVEN,
           tool_choice: dict | NotGiven = NOT_GIVEN,
           tools: Iterable[ToolParam] | NotGiven = NOT_GIVEN,
           top_k: int | NotGiven = NOT_GIVEN,
           top_p: float | NotGiven = NOT_GIVEN,
           extra_headers: Headers | None = None,
           extra_query: Query | None = None,
           extra_body: Body | None = None,
           timeout: float | httpx.Timeout | None | NotGiven = NOT_GIVEN,
           **kwargs) -> Any

Create a message with automatic streaming detection and Maxim logging.

This is the main entry point for message creation. It automatically detects whether streaming is requested and routes to the appropriate handler while ensuring all interactions are logged through Maxim.

Arguments:

Returns:

Notes:

The method automatically detects streaming mode via the ‘stream’ parameter in kwargs and routes accordingly.

​

stream

def stream(*args, **kwargs) -> Any

Create a streaming message with Maxim logging.

This method is a direct alias for create_stream, providing compatibility with the standard Anthropic Messages API while adding Maxim logging.

Arguments:

Returns: