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:
NameTypeDescription
_loggerLoggerThe Maxim logger instance for tracking interactions.

​
__init__

def __init__(client: Anthropic, logger: Logger)
Initialize the Maxim Anthropic Messages client. Arguments:
NameTypeDescription
clientAnthropicThe Anthropic client instance.
loggerLoggerThe Maxim logger instance for tracking and
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:
NameDescription
*argsVariable length argument list passed to the parent create method.
**kwargsArbitrary keyword arguments passed to the parent create method.
Special headers:
  • x-maxim-trace-id: Optional trace ID for associating with existing trace.
  • x-maxim-generation-name: Optional name for the generation.
Returns:
NameDescription
AnyThe response from the Anthropic API create method.
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:
NameDescription
*argsVariable length argument list passed to the parent stream method.
**kwargsArbitrary keyword arguments passed to the parent stream method.
Special headers:
  • x-maxim-trace-id: Optional trace ID for associating with existing trace.
  • x-maxim-generation-name: Optional name for the generation.
Returns:
NameDescription
[StreamWrapper](/sdk/python/references/logger/anthropic/stream_manager)A wrapped stream manager that processes chunks and
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:
NameTypeDescription
max_tokensintThe maximum number of tokens to generate.
messagesIterable[MessageParam]The conversation messages.
modelstrThe model to use for generation.
metadata_MetadataParamNotGiven_Additional metadata for the request.
stop_sequences_List[str]NotGiven_Sequences that will stop generation.
system_Union[str, Iterable[TextBlockParam]]NotGiven_System message.
temperature_floatNotGiven_Sampling temperature (0-1).
tool_choice_dictNotGiven_How the model should use tools.
tools_Iterable[ToolParam]NotGiven_Available tools for the model.
top_k_intNotGiven_Top-k sampling parameter.
top_p_floatNotGiven_Top-p (nucleus) sampling parameter.
extra_headers_HeadersNone_Additional HTTP headers.
extra_query_QueryNone_Additional query parameters.
extra_body_BodyNone_Additional request body data.
timeout_floathttpx.TimeoutNoneNotGiven_Request timeout.
**kwargsAdditional arguments, including ‘stream’ for streaming mode.
Returns:
NameDescription
AnyEither a direct message response or a StreamWrapper for streaming.
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:
NameDescription
*argsVariable length argument list passed to create_stream.
**kwargsArbitrary keyword arguments passed to create_stream.
Returns:
NameDescription
[StreamWrapper](/sdk/python/references/logger/anthropic/stream_manager)A wrapped stream manager with logging capabilities.