diff --git a/autogen/llm_clients/__init__.py b/autogen/llm_clients/__init__.py index 402ff1a28ef..7dca03cc598 100644 --- a/autogen/llm_clients/__init__.py +++ b/autogen/llm_clients/__init__.py @@ -34,6 +34,7 @@ ContentParser.register("custom_type", CustomContent) """ +from .anthropic_v2 import AnthropicV2Client, AnthropicV2LLMConfigEntry from .client_v2 import ModelClientV2 from .models import ( AudioContent, @@ -58,6 +59,7 @@ "ModelClientV2", # Clients "OpenAICompletionsClient", + "AnthropicV2Client", # Content blocks "AudioContent", "BaseContent", @@ -74,4 +76,6 @@ # Unified formats "UnifiedMessage", "UnifiedResponse", + # Config Entry + "AnthropicV2LLMConfigEntry", ] diff --git a/autogen/llm_clients/anthropic_v2.py b/autogen/llm_clients/anthropic_v2.py new file mode 100644 index 00000000000..752f53a646c --- /dev/null +++ b/autogen/llm_clients/anthropic_v2.py @@ -0,0 +1,1109 @@ +# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors +# +# SPDX-License-Identifier: Apache-2.0 + +""" +Anthropic Messages API Client implementing ModelClientV2 and ModelClient protocols. + +This client handles the Anthropic Messages API (client.messages.create) which returns +rich responses with: +- Thinking blocks (extended thinking feature) +- Tool calls and function execution +- Native structured outputs (beta API) +- JSON Mode structured outputs (fallback) +- Standard chat messages + +The client preserves all provider-specific features in UnifiedResponse format +and is compatible with AG2's agent system through ModelClient protocol. + +Note: This uses the Messages API, supporting both standard and beta structured outputs. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import time +import warnings +from typing import Any, Literal + +from pydantic import BaseModel + +from autogen.import_utils import optional_import_block + +logger = logging.getLogger(__name__) + +# Import Anthropic SDK with optional import handling +with optional_import_block() as anthropic_result: + from anthropic import Anthropic, BadRequestError + from anthropic.types import Message + + # Beta imports for structured outputs + try: + from anthropic import transform_schema + except ImportError: + transform_schema = None # type: ignore[misc, assignment] + + +if anthropic_result.is_successful: + anthropic_import_exception: ImportError | None = None +else: + Anthropic = None # type: ignore[assignment] + BadRequestError = None # type: ignore[assignment] + Message = None # type: ignore[assignment] + anthropic_import_exception = ImportError( + "Please install anthropic to use AnthropicCompletionsClient. Install with: pip install anthropic" + ) + +# Import helper functions and constants from existing anthropic.py +from autogen.oai.anthropic import ( + AnthropicEntryDict, + AnthropicLLMConfigEntry, + _calculate_cost, + _is_text_block, + _is_thinking_block, + _is_tool_use_block, + has_beta_messages_api, + oai_messages_to_anthropic_messages, + supports_native_structured_outputs, + transform_schema_for_anthropic, + validate_structured_outputs_version, +) +from autogen.oai.client_utils import FormatterProtocol, validate_parameter + +# Import for backward compatibility +from autogen.oai.oai_models import ( + ChatCompletion, + ChatCompletionMessage, + ChatCompletionMessageToolCall, + Choice, + CompletionUsage, +) + +# Import ModelClient protocol +from ..llm_config.client import ModelClient + +# Import UnifiedResponse models +from .models import ( + AudioContent, + GenericContent, + ImageContent, + ReasoningContent, + TextContent, + ToolCallContent, + UnifiedMessage, + UnifiedResponse, + VideoContent, + normalize_role, +) + + +class AnthropicV2LLMConfigDict(AnthropicEntryDict, total=False): + api_type: Literal["anthropic_v2"] + + +class AnthropicV2LLMConfigEntry(AnthropicLLMConfigEntry): + """ + LLMConfig entry for Anthropic V2 Client with ModelClientV2 architecture. + + This uses the new AnthropicV2Client from autogen.llm_clients which returns + rich UnifiedResponse objects with typed content blocks (ReasoningContent, + CitationContent, ToolCallContent, etc.). + """ + + api_type: Literal["anthropic_v2"] = "anthropic_v2" + + +class AnthropicV2Client(ModelClient): + """ + Anthropic Messages API client implementing ModelClientV2 protocol. + + This client works with Anthropic's Messages API (client.messages.create) + which returns structured output with thinking blocks, tool calls, and more. + + Key Features: + - Preserves thinking blocks as ReasoningContent (extended thinking feature) + - Handles tool calls and results + - Supports native structured outputs (beta API) and JSON Mode fallback + - Provides backward compatibility via create_v1_compatible() + - Supports multiple authentication methods (API key, AWS Bedrock, GCP Vertex) + + Example: + client = AnthropicCompletionsClient(api_key="...") + + # Get rich response with thinking + response = client.create({ + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "Explain quantum computing"}] + }) + + # Access thinking blocks + for reasoning in response.reasoning: + print(f"Thinking: {reasoning.reasoning}") + + # Get text response + print(f"Answer: {response.text}") + """ + + RESPONSE_USAGE_KEYS: list[str] = ["prompt_tokens", "completion_tokens", "total_tokens", "cost", "model"] + + def __init__( + self, + api_key: str | None = None, + base_url: str | None = None, + timeout: int | None = None, + response_format: type[BaseModel] | dict | None = None, + **kwargs: Any, + ): + """ + Initialize Anthropic Messages API client. + + Args: + api_key: Anthropic API key (or set ANTHROPIC_API_KEY env var) + base_url: Optional base URL for the API + timeout: Optional timeout in seconds + response_format: Optional response format for structured outputs + **kwargs: Additional arguments passed to Anthropic client + """ + if anthropic_import_exception is not None: + raise anthropic_import_exception + + # Store credentials + self._api_key = api_key or os.getenv("ANTHROPIC_API_KEY") + + # Validate credentials + if self._api_key is None: + raise ValueError( + "API key is required to use the Anthropic API. Set api_key parameter or ANTHROPIC_API_KEY environment variable." + ) + + # Initialize Anthropic client + client_kwargs = {"api_key": self._api_key} + if base_url: + client_kwargs["base_url"] = base_url + if timeout: + client_kwargs["timeout"] = timeout + self._client = Anthropic(**client_kwargs, **kwargs) # type: ignore[misc] + + # Store response format for structured outputs + self._response_format: type[BaseModel] | dict | None = response_format + + def create(self, params: dict[str, Any]) -> UnifiedResponse: # type: ignore[override] + """ + Create a completion and return UnifiedResponse with all features preserved. + + This method implements ModelClient.create() but returns UnifiedResponse instead + of ModelClientResponseProtocol. The rich UnifiedResponse structure is compatible + via duck typing - it has .model attribute and works with message_retrieval(). + + Automatically selects the best structured output method: + - Native structured outputs for Claude Sonnet 4.5+ (guaranteed schema compliance) + - JSON Mode for older models (prompt-based with tags) + - Standard completion for requests without response_format + + Args: + params: Request parameters including: + - model: Model name (e.g., "claude-3-5-sonnet-20241022") + - messages: List of message dicts + - temperature: Optional temperature + - max_tokens: Optional max completion tokens + - tools: Optional tool definitions + - response_format: Optional Pydantic BaseModel or JSON schema dict + - **other Anthropic parameters + + Returns: + UnifiedResponse with thinking blocks, tool calls, and all content preserved + """ + model = params.get("model") + response_format = params.get("response_format") or self._response_format + + # Route to appropriate implementation based on model and response_format + if response_format: + self._response_format = response_format + params["response_format"] = response_format + + # Try native structured outputs if model supports it + if supports_native_structured_outputs(model) and has_beta_messages_api(): + try: + return self._create_with_native_structured_output(params) + except (BadRequestError, AttributeError, ValueError) as e: # type: ignore[misc] + # Fallback to JSON Mode if native API not supported or schema invalid + self._log_structured_output_fallback(e, model, response_format, params) + return self._create_with_json_mode(params) + else: + # Use JSON Mode for older models or when beta API unavailable + return self._create_with_json_mode(params) + else: + # Standard completion without structured outputs + return self._create_standard(params) + + def _create_standard(self, params: dict[str, Any]) -> UnifiedResponse: + """ + Create a standard completion without structured outputs. + + Args: + params: Request parameters + + Returns: + UnifiedResponse with all content blocks properly typed + """ + # Convert OAI messages to Anthropic format + anthropic_messages = oai_messages_to_anthropic_messages(params) + + # Prepare Anthropic API parameters using helper (handles tool conversion, None removal, etc.) + anthropic_params = self._prepare_anthropic_params(params, anthropic_messages) + + # Check if any tools use strict mode (requires beta API) + has_strict_tools = any(tool.get("strict") for tool in anthropic_params.get("tools", [])) + + if has_strict_tools: + # Validate SDK version supports structured outputs beta + validate_structured_outputs_version() + # Use beta API for strict tools + anthropic_params["betas"] = ["structured-outputs-2025-11-13"] + response = self._client.beta.messages.create(**anthropic_params) # type: ignore[misc] + else: + # Standard API for legacy tools + response = self._client.messages.create(**anthropic_params) # type: ignore[misc] + + # Transform to UnifiedResponse + return self._transform_response(response, anthropic_params["model"], anthropic_params) + + def _create_with_native_structured_output(self, params: dict[str, Any]) -> UnifiedResponse: + """ + Create completion using native structured outputs (beta API). + + This method uses Anthropic's beta structured outputs feature for guaranteed + schema compliance via constrained decoding. + + Args: + params: Request parameters + + Returns: + UnifiedResponse with structured JSON output + + Raises: + AttributeError: If SDK doesn't support beta API + Exception: If native structured output fails + """ + # Check if Anthropic's transform_schema is available + if transform_schema is None: + raise ImportError("Anthropic transform_schema not available. Please upgrade to anthropic>=0.74.1") + + # Get schema from response_format and transform it using Anthropic's function + if isinstance(self._response_format, type) and issubclass(self._response_format, BaseModel): + # For Pydantic models, use Anthropic's transform_schema directly + transformed_schema = transform_schema(self._response_format) + elif isinstance(self._response_format, dict): + # For dict schemas, use as-is (already in correct format) + schema = self._response_format + # Still apply our transformation for additionalProperties + transformed_schema = transform_schema_for_anthropic(schema) + else: + raise ValueError(f"Invalid response format: {self._response_format}") + + # Convert AG2 messages to Anthropic messages + anthropic_messages = oai_messages_to_anthropic_messages(params) + + # Prepare Anthropic API parameters using helper + anthropic_params = self._prepare_anthropic_params(params, anthropic_messages) + + # Validate SDK version supports structured outputs beta + validate_structured_outputs_version() + + # Add native structured output parameters + anthropic_params["betas"] = ["structured-outputs-2025-11-13"] + + # Use beta API + if not hasattr(self._client, "beta"): + raise AttributeError( + "Anthropic SDK does not support beta.messages API. Please upgrade to anthropic>=0.39.0" + ) + + # When both tools and structured output are configured, must use create() (not parse()) + # parse() doesn't support tools, so we convert Pydantic models to dict schemas + has_tools = "tools" in anthropic_params and anthropic_params["tools"] + + if has_tools or isinstance(self._response_format, dict): + # Use create() with output_format for: + # 1. Dict schemas (always) + # 2. Pydantic models when tools are present (parse() doesn't support tools) + anthropic_params["output_format"] = { + "type": "json_schema", + "schema": transformed_schema, + } + response = self._client.beta.messages.create(**anthropic_params) # type: ignore[misc] + else: + # Pydantic model without tools - use parse() for automatic validation + # parse() provides parsed_output attribute for direct model access + anthropic_params["output_format"] = self._response_format + response = self._client.beta.messages.parse(**anthropic_params) # type: ignore[misc] + + # Transform to UnifiedResponse with is_native_structured_output=True + return self._transform_response( + response, anthropic_params["model"], anthropic_params, is_native_structured_output=True + ) + + def _create_with_json_mode(self, params: dict[str, Any]) -> UnifiedResponse: + """ + Create completion using legacy JSON Mode with tags. + + This method uses prompt-based structured outputs for older Claude models + that don't support native structured outputs. + + Args: + params: Request parameters + + Returns: + UnifiedResponse with JSON output extracted from tags + """ + # Add response format instructions to system message before message conversion + self._add_response_format_to_system(params) + + # Convert AG2 messages to Anthropic messages + anthropic_messages = oai_messages_to_anthropic_messages(params) + + # Prepare Anthropic API parameters using helper + anthropic_params = self._prepare_anthropic_params(params, anthropic_messages) + + # Call Anthropic API + response = self._client.messages.create(**anthropic_params) # type: ignore[misc] + + # Extract JSON from tags + parsed_response = self._extract_json_response(response) + + # Transform to UnifiedResponse + unified_response = self._transform_response(response, anthropic_params["model"], anthropic_params) + + # Replace text content with parsed JSON if structured output + if self._response_format: + # Find and replace TextContent with parsed JSON + for msg in unified_response.messages: + for i, block in enumerate(msg.content): + if isinstance(block, TextContent): + # Replace with parsed JSON text + json_text = ( + parsed_response.model_dump_json() + if hasattr(parsed_response, "model_dump_json") + else str(parsed_response) + ) + msg.content[i] = TextContent(type="text", text=json_text) + break + + return unified_response + + def _transform_response( + self, + anthropic_response: Message, # type: ignore[valid-type] + model: str, + anthropic_params: dict[str, Any], + is_native_structured_output: bool = False, + ) -> UnifiedResponse: + """ + Transform Anthropic Message response to UnifiedResponse. + + Handles all Anthropic content types: + - Text blocks → TextContent + - Thinking blocks → ReasoningContent + - Tool use blocks → ToolCallContent + - Structured outputs (parsed_output) → GenericContent with 'parsed' type + - Structured outputs from .create() → Parsed JSON into Pydantic model + - Unknown fields → GenericContent (forward compatibility) + + Args: + anthropic_response: Raw Anthropic Message response + model: Model name + anthropic_params: Original request parameters (needed for response_format access) + is_native_structured_output: Whether this is a native structured output response + + Returns: + UnifiedResponse with all content blocks properly typed + """ + content_blocks = [] + + # Process all content blocks from Anthropic response + for block in anthropic_response.content: + # Extract thinking content (extended thinking feature) + if _is_thinking_block(block): + content_blocks.append( + ReasoningContent( + type="reasoning", + reasoning=block.thinking, + summary=None, + ) + ) + # Extract tool calls (handles both ToolUseBlock and BetaToolUseBlock) + elif _is_tool_use_block(block): + content_blocks.append( + ToolCallContent( + type="tool_call", + id=block.id, + name=block.name, + arguments=json.dumps(block.input), + ) + ) + # Extract text content (handles both TextBlock and BetaTextBlock) + elif _is_text_block(block): + # For native structured output, handle both .parse() and .create() responses + if is_native_structured_output: + # Check if we have parsed_output (from .parse()) + if hasattr(anthropic_response, "parsed_output") and anthropic_response.parsed_output is not None: + parsed_response = anthropic_response.parsed_output + # Store parsed object as GenericContent to preserve it + if hasattr(parsed_response, "model_dump"): + parsed_dict = parsed_response.model_dump() + elif hasattr(parsed_response, "dict"): + parsed_dict = parsed_response.dict() + else: + parsed_dict = {"value": str(parsed_response)} + + content_blocks.append(GenericContent(type="parsed", parsed=parsed_dict)) + + # Also add text representation + text_content = ( + parsed_response.model_dump_json() + if hasattr(parsed_response, "model_dump_json") + else str(parsed_response) + ) + content_blocks.append(TextContent(type="text", text=text_content)) + else: + # Using .create() - parse JSON text into Pydantic model if available + # Check if we have a Pydantic model to parse into + if ( + self._response_format + and isinstance(self._response_format, type) + and issubclass(self._response_format, BaseModel) + ): + try: + # Parse JSON string into Pydantic model + json_data = json.loads(block.text) + parsed_response = self._response_format.model_validate(json_data) + + # Store parsed object as GenericContent + parsed_dict = parsed_response.model_dump() + content_blocks.append(GenericContent(type="parsed", parsed=parsed_dict)) + + # Add text representation + text_content = parsed_response.model_dump_json() + content_blocks.append(TextContent(type="text", text=text_content)) + except (json.JSONDecodeError, ValueError) as e: + # If parsing fails, log warning and use text as-is + logger.warning(f"Failed to parse structured output JSON: {e}") + content_blocks.append(TextContent(type="text", text=block.text)) + else: + # Dict schema or no model - just use text as-is + content_blocks.append(TextContent(type="text", text=block.text)) + else: + # Regular text content (not structured output) + content_blocks.append(TextContent(type="text", text=block.text)) + + # Fallback: If using native SO parse() and no content blocks were found, + # extract from parsed_output directly + if ( + not content_blocks + and is_native_structured_output + and hasattr(anthropic_response, "parsed_output") + and anthropic_response.parsed_output is not None + ): + parsed_response = anthropic_response.parsed_output + # Store parsed object as GenericContent + if hasattr(parsed_response, "model_dump"): + parsed_dict = parsed_response.model_dump() + elif hasattr(parsed_response, "dict"): + parsed_dict = parsed_response.dict() + else: + parsed_dict = {"value": str(parsed_response)} + + content_blocks.append(GenericContent(type="parsed", parsed=parsed_dict)) + + # Add text representation + text_content = ( + parsed_response.model_dump_json() + if hasattr(parsed_response, "model_dump_json") + else str(parsed_response) + ) + content_blocks.append(TextContent(type="text", text=text_content)) + + # Create unified message with normalized role (Anthropic responses are always assistant) + messages = [ + UnifiedMessage( + role=normalize_role("assistant"), + content=content_blocks, + ) + ] + + # Extract usage information + usage = { + "prompt_tokens": anthropic_response.usage.input_tokens, + "completion_tokens": anthropic_response.usage.output_tokens, + "total_tokens": anthropic_response.usage.input_tokens + anthropic_response.usage.output_tokens, + } + + # Determine finish reason + finish_reason = "stop" + if anthropic_response.stop_reason == "tool_use": + finish_reason = "tool_calls" + + # Build UnifiedResponse + unified_response = UnifiedResponse( + id=anthropic_response.id, + model=model, + provider="anthropic", + messages=messages, + usage=usage, + finish_reason=finish_reason, + status="completed", + provider_metadata={ + "stop_reason": anthropic_response.stop_reason, + "stop_sequence": getattr(anthropic_response, "stop_sequence", None), + }, + ) + + # Calculate cost + unified_response.cost = self.cost(unified_response) + + return unified_response + + def load_config(self, params: dict[str, Any]) -> dict[str, Any]: + """Load the configuration for the Anthropic API client.""" + anthropic_params = {} + + anthropic_params["model"] = params.get("model") + assert anthropic_params["model"], "Please provide a `model` in the config_list to use the Anthropic API." + + anthropic_params["temperature"] = validate_parameter( + params, "temperature", (float, int), False, 1.0, (0.0, 1.0), None + ) + anthropic_params["max_tokens"] = validate_parameter(params, "max_tokens", int, False, 4096, (1, None), None) + anthropic_params["timeout"] = validate_parameter(params, "timeout", int, True, None, (1, None), None) + anthropic_params["top_k"] = validate_parameter(params, "top_k", int, True, None, (1, None), None) + anthropic_params["top_p"] = validate_parameter(params, "top_p", (float, int), True, None, (0.0, 1.0), None) + anthropic_params["stop_sequences"] = validate_parameter(params, "stop_sequences", list, True, None, None, None) + anthropic_params["stream"] = validate_parameter(params, "stream", bool, False, False, None, None) + if "thinking" in params: + anthropic_params["thinking"] = params["thinking"] + + if anthropic_params["stream"]: + warnings.warn( + "Streaming is not currently supported, streaming will be disabled.", + UserWarning, + ) + anthropic_params["stream"] = False + + # Note the Anthropic API supports "tool" for tool_choice but you must specify the tool name so we will ignore that here + # Dictionary, see options here: https://docs.anthropic.com/en/docs/build-with-claude/tool-use/overview#controlling-claudes-output + # type = auto, any, tool, none | name = the name of the tool if type=tool + anthropic_params["tool_choice"] = validate_parameter(params, "tool_choice", dict, True, None, None, None) + + return anthropic_params + + def _remove_none_params(self, params: dict[str, Any]) -> None: + """Remove parameters with None values from the params dict. + + Anthropic API doesn't accept None values, so we remove them before making requests. + This method modifies the params dict in-place. + + Args: + params: Dictionary of API parameters + """ + keys_to_remove = [key for key, value in params.items() if value is None] + for key in keys_to_remove: + del params[key] + + def _prepare_anthropic_params( + self, params: dict[str, Any], anthropic_messages: list[dict[str, Any]] + ) -> dict[str, Any]: + """ + Prepare parameters for Anthropic API call. + + Consolidates common parameter preparation logic used across all create methods: + - Loads base configuration + - Converts tools format if needed + - Assigns messages, system, and tools + - Removes None values + + Args: + params: Original request parameters + anthropic_messages: Converted messages in Anthropic format + + Returns: + Dictionary of Anthropic API parameters ready for use + """ + # Load base configuration + anthropic_params = self.load_config(params) + + # Convert tools to functions if needed (make a copy to avoid modifying original) + params_copy = params.copy() + if "functions" in params_copy: + tools_configs = params_copy.pop("functions") + tools_configs = [self.openai_func_to_anthropic(tool) for tool in tools_configs] + params_copy["tools"] = tools_configs + elif "tools" in params_copy: + # Convert OpenAI tool format to Anthropic format + # OpenAI format: {"type": "function", "function": {...}} + # Anthropic format: {"name": "...", "description": "...", "input_schema": {...}} + tools_configs = self.convert_tools_to_functions(params_copy.pop("tools")) + tools_configs = [self.openai_func_to_anthropic(tool) for tool in tools_configs] + params_copy["tools"] = tools_configs + + # Assign messages and optional parameters + anthropic_params["messages"] = anthropic_messages + if "system" in params_copy: + anthropic_params["system"] = params_copy["system"] + if "tools" in params_copy: + anthropic_params["tools"] = params_copy["tools"] + + # Remove None values + self._remove_none_params(anthropic_params) + + return anthropic_params + + def _extract_json_response(self, response: Message) -> Any: # type: ignore[valid-type] + """ + Extract and validate JSON response from the output for structured outputs. + + Args: + response: The response from the API + + Returns: + The parsed JSON response + """ + if not self._response_format: + return response + + # Extract content from response - check both thinking and text blocks + content = "" + if response.content: + for block in response.content: + if _is_thinking_block(block): + content = block.thinking + break + elif _is_text_block(block): + content = block.text + break + + # Try to extract JSON from tags first + json_match = re.search(r"(.*?)", content, re.DOTALL) + if json_match: + json_str = json_match.group(1).strip() + else: + # Fallback to finding first JSON object + json_start = content.find("{") + json_end = content.rfind("}") + if json_start == -1 or json_end == -1: + raise ValueError("No valid JSON found in response for Structured Output.") + json_str = content[json_start : json_end + 1] + + try: + # Parse JSON and validate against the Pydantic model if Pydantic model was provided + json_data = json.loads(json_str) + if isinstance(self._response_format, dict): + return json_str + else: + return self._response_format.model_validate(json_data) + + except Exception as e: + raise ValueError(f"Failed to parse response as valid JSON matching the schema for Structured Output: {e!s}") + + def _resolve_schema_refs(self, schema: dict[str, Any], defs: dict[str, Any]) -> dict[str, Any]: + """Recursively resolve $ref references in a JSON schema. + + Args: + schema: The schema to resolve + defs: The definitions dict from $defs + + Returns: + Schema with all $ref references resolved inline + """ + if isinstance(schema, dict): + if "$ref" in schema: + # Extract the reference name (e.g., "#/$defs/Step" -> "Step") + ref_name = schema["$ref"].split("/")[-1] + # Replace with the actual definition + return self._resolve_schema_refs(defs[ref_name].copy(), defs) + else: + # Recursively resolve all nested schemas + return {k: self._resolve_schema_refs(v, defs) for k, v in schema.items()} + elif isinstance(schema, list): + return [self._resolve_schema_refs(item, defs) for item in schema] + else: + return schema + + def _add_response_format_to_system(self, params: dict[str, Any]) -> None: + """ + Add prompt that will generate properly formatted JSON for structured outputs to system parameter. + + Based on Anthropic's JSON Mode cookbook, we ask the LLM to put the JSON within tags. + + Args: + params: The client parameters (modified in place) + """ + # Get the schema of the Pydantic model + if isinstance(self._response_format, dict): + schema = self._response_format + else: + # Use mode='serialization' and ref_template='{model}' to get a flatter, more LLM-friendly schema + schema = self._response_format.model_json_schema(mode="serialization", ref_template="{model}") + + # Resolve $ref references for simpler schema + if "$defs" in schema: + defs = schema.pop("$defs") + schema = self._resolve_schema_refs(schema, defs) + + # Add instructions for JSON formatting + # Generate an example based on the actual schema + def generate_example(schema_dict: dict[str, Any]) -> dict[str, Any]: + """Generate example data from schema.""" + example = {} + properties = schema_dict.get("properties", {}) + for prop_name, prop_schema in properties.items(): + prop_type = prop_schema.get("type", "string") + if prop_type == "string": + example[prop_name] = f"example {prop_name}" + elif prop_type == "integer": + example[prop_name] = 42 + elif prop_type == "number": + example[prop_name] = 42.0 + elif prop_type == "boolean": + example[prop_name] = True + elif prop_type == "array": + items_schema = prop_schema.get("items", {}) + items_type = items_schema.get("type", "string") + if items_type == "string": + example[prop_name] = ["item1", "item2"] + elif items_type == "object": + example[prop_name] = [generate_example(items_schema)] + else: + example[prop_name] = [] + elif prop_type == "object": + example[prop_name] = generate_example(prop_schema) + else: + example[prop_name] = f"example {prop_name}" + return example + + example_data = generate_example(schema) + example_json = json.dumps(example_data, indent=2) + + format_content = f"""You must respond with a valid JSON object that matches this structure (do NOT return the schema itself): +{json.dumps(schema, indent=2)} + +IMPORTANT: Put your actual response data (not the schema) inside tags. + +Correct example format: + +{example_json} + + +WRONG: Do not return the schema definition itself. + +Your JSON must: +1. Match the schema structure above +2. Contain actual data values, not schema descriptions +3. Be valid, parseable JSON""" + + # Add formatting to system message (create one if it doesn't exist) + if "system" in params: + params["system"] = params["system"] + "\n\n" + format_content + else: + params["system"] = format_content + + def _log_structured_output_fallback( + self, + exception: Exception, + model: str | None, + response_format: Any, + params: dict[str, Any], + ) -> None: + """ + Log detailed error information when native structured output fails and we fallback to JSON Mode. + + Args: + exception: The exception that triggered the fallback + model: Model name/identifier + response_format: Response format specification (Pydantic model or dict) + params: Original request parameters + """ + # Build error details dictionary + error_details = { + "model": model, + "response_format": str( + type(response_format).__name__ if isinstance(response_format, type) else type(response_format) + ), + "error_type": type(exception).__name__, + "error_message": str(exception), + } + + # Add BadRequestError-specific details if available + if isinstance(exception, BadRequestError): # type: ignore[misc] + if hasattr(exception, "status_code"): + error_details["status_code"] = exception.status_code + if hasattr(exception, "response"): + error_details["response_body"] = str( + exception.response.text if hasattr(exception.response, "text") else exception.response + ) + if hasattr(exception, "body"): + error_details["error_body"] = str(exception.body) + + # Log sanitized params (remove sensitive data like API keys, message content) + sanitized_params = { + "model": params.get("model"), + "max_tokens": params.get("max_tokens"), + "temperature": params.get("temperature"), + "has_tools": "tools" in params, + "num_messages": len(params.get("messages", [])), + } + error_details["params"] = sanitized_params + + # Log warning with full error context + logger.warning( + f"Native structured output failed for {model}. Error: {error_details}. Falling back to JSON Mode." + ) + + def create_v1_compatible(self, params: dict[str, Any]) -> ChatCompletion: + """ + Create completion in backward-compatible ChatCompletion format. + + This method provides compatibility with existing AG2 code that expects + ChatCompletion format. Note that thinking blocks will be preserved in + the content string with [Thinking] tags, matching V1 behavior. + + Args: + params: Same parameters as create() + + Returns: + ChatCompletion object compatible with OpenAI format + + Warning: + This method may lose some information when converting to the legacy format. + Prefer create() for new code. + """ + # Get rich response + unified_response = self.create(params) + + # Build message text with proper thinking block formatting (matching V1 behavior) + message_text = "" + for msg in unified_response.messages: + # Extract reasoning blocks (thinking content) + reasoning_blocks = msg.get_reasoning() + # Extract text content blocks + text_blocks = [b for b in msg.content if isinstance(b, TextContent)] + + # Combine thinking content (multiple blocks joined with \n\n) + thinking_content = "\n\n".join([r.reasoning for r in reasoning_blocks]) + # Combine text content (multiple blocks joined with \n\n) + text_content = "\n\n".join([t.text for t in text_blocks]) + + # Format like V1: [Thinking]\n{thinking}\n\n{text} + if thinking_content and text_content: + message_text = f"[Thinking]\n{thinking_content}\n\n{text_content}" + elif thinking_content: + message_text = f"[Thinking]\n{thinking_content}" + elif text_content: + message_text = text_content + break # Anthropic responses have single message + + # Extract tool calls if present + tool_calls = None + for msg in unified_response.messages: + tool_call_blocks = msg.get_tool_calls() + if tool_call_blocks: + tool_calls = [ + ChatCompletionMessageToolCall( + id=tc.id, + function={"name": tc.name, "arguments": tc.arguments}, + type="function", + ) + for tc in tool_call_blocks + ] + break + + # Build ChatCompletion + message = ChatCompletionMessage( + role="assistant", + content=message_text, + function_call=None, + tool_calls=tool_calls, + ) + + choices = [Choice(finish_reason=unified_response.finish_reason or "stop", index=0, message=message)] + + return ChatCompletion( + id=unified_response.id, + model=unified_response.model, + created=int(time.time()), + object="chat.completion", + choices=choices, + usage=CompletionUsage( + prompt_tokens=unified_response.usage.get("prompt_tokens", 0), + completion_tokens=unified_response.usage.get("completion_tokens", 0), + total_tokens=unified_response.usage.get("total_tokens", 0), + ), + cost=unified_response.cost or 0.0, + ) + + def message_retrieval(self, response: UnifiedResponse) -> list[str] | list[ChatCompletionMessage]: # type: ignore[override] + """ + Retrieve messages from response in OpenAI-compatible format. + + Returns list of strings for text-only messages, or list of dicts when + tool calls or complex content is present. + + Args: + response: UnifiedResponse from create() + + Returns: + List of strings (for text-only) OR list of message dicts (for tool calls/complex content) + """ + result: list[str] | list[ChatCompletionMessage] = [] + + for msg in response.messages: + # Check for tool calls + tool_calls = msg.get_tool_calls() + + # Check for complex/multimodal content that needs dict format + has_complex_content = any( + isinstance(block, (ImageContent, AudioContent, VideoContent)) for block in msg.content + ) + + if tool_calls or has_complex_content: + # Return OpenAI-compatible dict format + message_dict = ChatCompletionMessage( + role=msg.role.value if hasattr(msg.role, "value") else msg.role, + content=msg.get_text() or None, + ) + + # Add tool calls in OpenAI format + if tool_calls: + message_dict.tool_calls = [ + ChatCompletionMessageToolCall( + id=tc.id, + type="function", + function={"name": tc.name, "arguments": tc.arguments}, + ) + for tc in tool_calls + ] + + result.append(message_dict) + else: + # Simple text content - apply FormatterProtocol if available + content = msg.get_text() + + # If response_format implements FormatterProtocol (has format() method), use it + if isinstance(self._response_format, FormatterProtocol): + try: + # Try to parse and format + parsed = self._response_format.model_validate_json(content) # type: ignore[union-attr] + content = parsed.format() # type: ignore[union-attr] + except Exception: + # If parsing fails, return as-is + pass + + result.append(content) + + return result + + def cost(self, response: UnifiedResponse) -> float: # type: ignore[override] + """ + Calculate cost from response usage. + + Implements ModelClient.cost() but accepts UnifiedResponse via duck typing. + + Args: + response: UnifiedResponse with usage information + + Returns: + Cost in USD for the API call + """ + if not response.usage: + return 0.0 + + model = response.model + prompt_tokens = response.usage.get("prompt_tokens", 0) + completion_tokens = response.usage.get("completion_tokens", 0) + + return _calculate_cost(prompt_tokens, completion_tokens, model) + + @staticmethod + def get_usage(response: UnifiedResponse) -> dict[str, Any]: # type: ignore[override] + """ + Extract usage statistics from response. + + Implements ModelClient.get_usage() but accepts UnifiedResponse via duck typing. + + Args: + response: UnifiedResponse from create() + + Returns: + Dict with keys from RESPONSE_USAGE_KEYS + """ + return { + "prompt_tokens": response.usage.get("prompt_tokens", 0), + "completion_tokens": response.usage.get("completion_tokens", 0), + "total_tokens": response.usage.get("total_tokens", 0), + "cost": response.cost or 0.0, + "model": response.model, + } + + @staticmethod + def openai_func_to_anthropic(openai_func: dict) -> dict: + """Convert OpenAI function format to Anthropic format. + + Args: + openai_func: OpenAI function definition + + Returns: + Anthropic function definition + """ + res = openai_func.copy() + res["input_schema"] = res.pop("parameters") + + # Preserve strict field if present (for Anthropic structured outputs) + # strict=True enables guaranteed schema validation for tool inputs + if "strict" in openai_func: + res["strict"] = openai_func["strict"] + # Transform schema to add required additionalProperties: false for all objects + # Anthropic requires this for strict tools + res["input_schema"] = transform_schema_for_anthropic(res["input_schema"]) + + return res + + @staticmethod + def convert_tools_to_functions(tools: list) -> list: + """Convert tool definitions into Anthropic-compatible functions, + updating nested $ref paths in property schemas. + + Args: + tools: List of tool definitions + + Returns: + List of functions with updated $ref paths + """ + + def update_refs(obj: Any, defs_keys: set[str], prop_name: str) -> None: + """Recursively update $ref values that start with "#/$defs/".""" + if isinstance(obj, dict): + for key, value in obj.items(): + if key == "$ref" and isinstance(value, str) and value.startswith("#/$defs/"): + ref_key = value[len("#/$defs/") :] + if ref_key in defs_keys: + obj[key] = f"#/properties/{prop_name}/$defs/{ref_key}" + else: + update_refs(value, defs_keys, prop_name) + elif isinstance(obj, list): + for item in obj: + update_refs(item, defs_keys, prop_name) + + functions = [] + for tool in tools: + if tool.get("type") == "function" and "function" in tool: + function = tool["function"] + parameters = function.get("parameters", {}) + properties = parameters.get("properties", {}) + for prop_name, prop_schema in properties.items(): + if "$defs" in prop_schema: + defs_keys = set(prop_schema["$defs"].keys()) + update_refs(prop_schema, defs_keys, prop_name) + functions.append(function) + return functions diff --git a/autogen/llm_config/types.py b/autogen/llm_config/types.py index fa74f511b2b..b6f968d4df8 100644 --- a/autogen/llm_config/types.py +++ b/autogen/llm_config/types.py @@ -2,6 +2,7 @@ # # SPDX-License-Identifier: Apache-2.0 +from autogen.llm_clients.anthropic_v2 import AnthropicV2LLMConfigEntry from autogen.oai.anthropic import AnthropicLLMConfigEntry from autogen.oai.bedrock import BedrockLLMConfigEntry from autogen.oai.cerebras import CerebrasLLMConfigEntry @@ -21,6 +22,7 @@ ConfigEntries = ( AnthropicLLMConfigEntry + | AnthropicV2LLMConfigEntry | CerebrasLLMConfigEntry | BedrockLLMConfigEntry | AzureOpenAILLMConfigEntry diff --git a/autogen/oai/anthropic.py b/autogen/oai/anthropic.py index cf4c57549b5..aeefc4cbdc4 100644 --- a/autogen/oai/anthropic.py +++ b/autogen/oai/anthropic.py @@ -1232,7 +1232,7 @@ def process_image_content(content_item: dict[str, Any]) -> dict[str, Any]: url = content_item["image_url"]["url"] try: - # Handle data URLs + # Handle data URLs (base64 encoded) if url.startswith("data:"): data_url_pattern = r"data:image/([a-zA-Z]+);base64,(.+)" match = re.match(data_url_pattern, url) @@ -1243,13 +1243,18 @@ def process_image_content(content_item: dict[str, Any]) -> dict[str, Any]: "source": {"type": "base64", "media_type": f"image/{media_type}", "data": base64_data}, } - else: - print("Error processing image.") - # Return original content if image processing fails - return content_item + # Handle regular HTTP/HTTPS URLs + elif url.startswith("http://") or url.startswith("https://"): + return { + "type": "image", + "source": {"type": "url", "url": url}, + } + + # If URL format is not recognized, return original content + return content_item except Exception as e: - print(f"Error processing image image: {e}") + print(f"Error processing image: {e}") # Return original content if image processing fails return content_item diff --git a/autogen/oai/client.py b/autogen/oai/client.py index 4f3d502410c..38b17ef5fca 100644 --- a/autogen/oai/client.py +++ b/autogen/oai/client.py @@ -1034,6 +1034,14 @@ def create_azure_openai_client() -> AzureOpenAI: ) self._clients.append(v2_client) # type: ignore[arg-type] client = v2_client + + elif api_type is not None and api_type.startswith("anthropic_v2"): + from autogen.llm_clients import AnthropicV2Client as V2Client + + v2_client = V2Client(response_format=response_format, **openai_config) + self._clients.append(v2_client) # type: ignore[arg-type] + client = v2_client + elif api_type is not None and api_type.startswith("responses"): # OpenAI Responses API (stateful). Reuse the same OpenAI SDK but call the `/responses` endpoint via the new client. @require_optional_import("openai>=1.66.2", "openai") diff --git a/notebook/agentchat_v2_anthropic_client_example.ipynb b/notebook/agentchat_v2_anthropic_client_example.ipynb new file mode 100644 index 00000000000..57856fcabeb --- /dev/null +++ b/notebook/agentchat_v2_anthropic_client_example.ipynb @@ -0,0 +1,628 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Anthropic V2 Client with AG2 Agents\n", + "\n", + "Author: [Priyanshu Deshmukh](https://github.com/priyansh4320)\n", + "\n", + "This notebook demonstrates how to use the **Anthropic V2 Client** (`api_type: \"anthropic_v2\"`) with AG2's agent system. The V2 client provides:\n", + "\n", + "- **Rich UnifiedResponse objects**: Typed content blocks (TextContent, ReasoningContent, ToolCallContent, etc.)\n", + "- **Structured Outputs**: Guaranteed schema-compliant JSON responses\n", + "- **Strict Tool Use**: Type-safe function calls with guaranteed schema validation\n", + "- **Vision Support**: Full multimodal capabilities with image input\n", + "- **Forward Compatibility**: GenericContent handles unknown future content types\n", + "\n", + "## What is Anthropic V2 Client?\n", + "\n", + "The V2 client implements `ModelClientV2` protocol, returning rich `UnifiedResponse` objects with:\n", + "\n", + "- **Typed content blocks**: `TextContent`, `ReasoningContent`, `ToolCallContent`, `CitationContent`\n", + "- **Structured outputs**: Native support for Pydantic models and JSON schemas\n", + "- **Strict tools**: Guaranteed schema validation for tool inputs\n", + "- **Rich metadata**: Full reasoning blocks, citations, and tool execution details\n", + "- **Type safety**: Pydantic validation for all response data\n", + "- **Cost tracking**: Automatic per-response cost calculation" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Installation\n", + "```bash\n", + "pip install ag2[anthrpic]\n", + "```" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Setup" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import os\n", + "import textwrap\n", + "\n", + "from dotenv import load_dotenv\n", + "from pydantic import BaseModel\n", + "\n", + "from autogen import AssistantAgent, UserProxyAgent\n", + "from autogen.io.run_response import Cost\n", + "\n", + "load_dotenv()\n", + "\n", + "\n", + "# Helper function to extract total cost from ChatResult.cost dict\n", + "def get_total_cost(cost_dict):\n", + " \"\"\"Extract total cost from ChatResult.cost dict structure.\"\"\"\n", + " total = 0.0\n", + " for usage_type in cost_dict.values():\n", + " if isinstance(usage_type, dict):\n", + " for model_usage in usage_type.values():\n", + " if isinstance(model_usage, dict) and \"cost\" in model_usage:\n", + " total += model_usage[\"cost\"]\n", + " return total\n", + "\n", + "\n", + "# Helper function to extract cost from run response\n", + "def get_total_cost_from_run(run_response_cost):\n", + " \"\"\"Extract total cost from run response object.\"\"\"\n", + " if isinstance(run_response_cost, Cost):\n", + " return run_response_cost.usage_including_cached_inference.total_cost\n", + " return 0.0\n", + "\n", + "\n", + "print(\"✅ Environment configured\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Part 1: Structured Outputs\n", + "\n", + "Anthropic's structured outputs feature provides two powerful modes:\n", + "\n", + "1. **JSON Outputs** (`response_format`): Get validated JSON responses matching a specific schema\n", + "2. **Strict Tool Use** (`strict: true`): Guaranteed schema validation for tool inputs\n", + "\n", + "### Key Benefits\n", + "\n", + "- **Always Valid**: No more `JSON.parse()` errors\n", + "- **Type Safe**: Guaranteed field types and required fields\n", + "- **Reliable**: No retries needed for schema violations\n", + "- **Dual Modes**: JSON for data extraction, strict tools for agentic workflows\n", + "\n", + "### Requirements\n", + "\n", + "- Claude Sonnet 4.5 (`claude-sonnet-4-5`) or Claude Opus 4.1 (`claude-opus-4-1`)\n", + "- Anthropic SDK >= 0.74.1\n", + "- Beta header: `structured-outputs-2025-11-13` (automatically applied by AG2)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Example 1: JSON Structured Outputs with Pydantic Models\n", + "\n", + "The most common use case is extracting structured data from unstructured text. We'll use Pydantic models to define our schema and get validated JSON responses.\n", + "\n", + "### Use Case: Mathematical Reasoning\n", + "\n", + "Let's create an agent that solves math problems and returns structured step-by-step reasoning." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Define the structured output schema using Pydantic\n", + "class Step(BaseModel):\n", + " \"\"\"A single step in mathematical reasoning.\"\"\"\n", + "\n", + " explanation: str\n", + " output: str\n", + "\n", + "\n", + "class MathReasoning(BaseModel):\n", + " \"\"\"Structured output for mathematical problem solving.\"\"\"\n", + "\n", + " steps: list[Step]\n", + " final_answer: str\n", + "\n", + " def format(self) -> str:\n", + " \"\"\"Format the response for display.\"\"\"\n", + " steps_output = \"\\n\".join(\n", + " f\"Step {i + 1}: {step.explanation}\\n Output: {step.output}\" for i, step in enumerate(self.steps)\n", + " )\n", + " return f\"{steps_output}\\n\\nFinal Answer: {self.final_answer}\"\n", + "\n", + "\n", + "# Configure LLM with structured output using V2 client\n", + "llm_config = {\n", + " \"config_list\": [\n", + " {\n", + " \"model\": \"claude-sonnet-4-5\",\n", + " \"api_key\": os.getenv(\"ANTHROPIC_API_KEY\"),\n", + " \"api_type\": \"anthropic_v2\", # <-- Use V2 client\n", + " \"response_format\": MathReasoning, # Enable structured outputs\n", + " }\n", + " ],\n", + "}\n", + "\n", + "# Create agents\n", + "user_proxy = UserProxyAgent(\n", + " name=\"User\",\n", + " human_input_mode=\"NEVER\",\n", + " max_consecutive_auto_reply=0,\n", + " code_execution_config=False,\n", + ")\n", + "\n", + "math_assistant = AssistantAgent(\n", + " name=\"MathAssistant\",\n", + " system_message=\"You are a math tutor. Solve problems step by step.\",\n", + " llm_config=llm_config,\n", + ")\n", + "\n", + "print(\"✅ Example 1 configured: Math reasoning with structured outputs\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Ask the assistant to solve a math problem\n", + "chat_result = user_proxy.run(\n", + " math_assistant,\n", + " message=\"Solve the equation: 3x + 7 = 22\",\n", + " max_turns=1,\n", + ")\n", + "\n", + "chat_result.process()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "### How It Works\n", + "\n", + "1. **Schema Definition**: Pydantic models define the expected structure\n", + "2. **Beta API**: AG2 automatically uses `beta.messages.parse()` for Pydantic models\n", + "3. **Constrained Decoding**: Claude generates output that strictly follows the schema\n", + "4. **FormatterProtocol**: If your model has a `format()` method, it's automatically called\n", + "\n", + "**Benefits**:\n", + "- ✅ No JSON parsing errors\n", + "- ✅ Guaranteed schema compliance\n", + "- ✅ Type-safe field access\n", + "- ✅ Custom formatting support" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Example 2: Strict Tool Use for Type-Safe Function Calls\n", + "\n", + "Strict tool use ensures that Claude's tool inputs exactly match your schema. This is critical for production agentic systems where invalid parameters can break workflows.\n", + "\n", + "### Use Case: Weather API with Validated Inputs\n", + "\n", + "Without strict mode, Claude might return `\"celsius\"` as a string when you expect an enum, or `\"2\"` instead of `2`. Strict mode guarantees correct types." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Define a tool function\n", + "def get_weather(location: str, unit: str = \"celsius\") -> str:\n", + " \"\"\"Get the weather for a location.\n", + "\n", + " Args:\n", + " location: The city and state, e.g. San Francisco, CA\n", + " unit: Temperature unit (celsius or fahrenheit)\n", + " \"\"\"\n", + " # In a real application, this would call a weather API\n", + " return f\"Weather in {location}: 22°{unit.upper()[0]}, partly cloudy\"\n", + "\n", + "\n", + "# Configure LLM with strict tool using V2 client\n", + "llm_config_strict = {\n", + " \"config_list\": [\n", + " {\n", + " \"model\": \"claude-sonnet-4-5\",\n", + " \"api_key\": os.getenv(\"ANTHROPIC_API_KEY\"),\n", + " \"api_type\": \"anthropic_v2\", # <-- Use V2 client\n", + " }\n", + " ],\n", + " \"functions\": [\n", + " {\n", + " \"name\": \"get_weather\",\n", + " \"description\": \"Get the weather for a location\",\n", + " \"strict\": True, # Enable strict schema validation ✨\n", + " \"parameters\": {\n", + " \"type\": \"object\",\n", + " \"properties\": {\n", + " \"location\": {\"type\": \"string\", \"description\": \"The city and state, e.g. San Francisco, CA\"},\n", + " \"unit\": {\"type\": \"string\", \"enum\": [\"celsius\", \"fahrenheit\"], \"description\": \"Temperature unit\"},\n", + " },\n", + " \"required\": [\"location\"],\n", + " },\n", + " }\n", + " ],\n", + "}\n", + "\n", + "# Create agents\n", + "weather_assistant = AssistantAgent(\n", + " name=\"WeatherAssistant\",\n", + " system_message=\"You help users get weather information. Use the get_weather function.\",\n", + " llm_config=llm_config_strict,\n", + ")\n", + "\n", + "user_proxy_2 = UserProxyAgent(\n", + " name=\"User\",\n", + " human_input_mode=\"NEVER\",\n", + " max_consecutive_auto_reply=1,\n", + " code_execution_config=False,\n", + ")\n", + "\n", + "# Register function on both agents\n", + "weather_assistant.register_function({\"get_weather\": get_weather})\n", + "user_proxy_2.register_function({\"get_weather\": get_weather})\n", + "\n", + "print(\"✅ Example 2 configured: Strict tool use for weather queries\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Query the weather\n", + "chat_result = user_proxy_2.initiate_chat(\n", + " weather_assistant,\n", + " message=\"What's the weather in Boston, MA?\",\n", + " max_turns=2,\n", + ")\n", + "\n", + "# Verify tool call had strict typing\n", + "print(\"\\n\" + \"=\" * 60)\n", + "print(\"TOOL CALL VERIFICATION:\")\n", + "print(\"=\" * 60)\n", + "\n", + "import json\n", + "\n", + "for message in chat_result.chat_history:\n", + " if message.get(\"tool_calls\"):\n", + " tool_call = message[\"tool_calls\"][0]\n", + " args = json.loads(tool_call[\"function\"][\"arguments\"])\n", + " print(f\"Function: {tool_call['function']['name']}\")\n", + " print(f\"Arguments: {args}\")\n", + " print(f\"✅ location type: {type(args['location']).__name__}\")\n", + " if \"unit\" in args:\n", + " print(f\"✅ unit value: {args['unit']} (valid enum)\")\n", + " break" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Example 3: Combined JSON Outputs + Strict Tools\n", + "\n", + "The most powerful pattern is combining both features: use strict tools for calculations/actions, then return structured JSON for the final result.\n", + "\n", + "### Use Case: Math Calculator Agent\n", + "\n", + "The agent uses strict tools to perform calculations (guaranteed correct types), then provides a structured summary of the work." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Define calculator tool\n", + "def calculate(operation: str, a: float, b: float) -> float:\n", + " \"\"\"Perform a calculation.\n", + "\n", + " Args:\n", + " operation: The operation to perform (add, subtract, multiply, divide)\n", + " a: First number\n", + " b: Second number\n", + " \"\"\"\n", + " if operation == \"add\":\n", + " return a + b\n", + " elif operation == \"subtract\":\n", + " return a - b\n", + " elif operation == \"multiply\":\n", + " return a * b\n", + " elif operation == \"divide\":\n", + " return a / b if b != 0 else 0\n", + " return 0\n", + "\n", + "\n", + "# Result model for structured output\n", + "class CalculationResult(BaseModel):\n", + " \"\"\"Structured output for calculation results.\"\"\"\n", + "\n", + " problem: str\n", + " steps: list[str]\n", + " result: float\n", + " verification: str\n", + "\n", + "\n", + "# Configure with BOTH features using V2 client\n", + "llm_config_combined = {\n", + " \"config_list\": [\n", + " {\n", + " \"model\": \"claude-sonnet-4-5\",\n", + " \"api_key\": os.getenv(\"ANTHROPIC_API_KEY\"),\n", + " \"api_type\": \"anthropic_v2\", # <-- Use V2 client\n", + " \"response_format\": CalculationResult, # 1. Structured JSON output\n", + " }\n", + " ],\n", + " \"functions\": [\n", + " { # 2. Strict tool validation\n", + " \"name\": \"calculate\",\n", + " \"description\": \"Perform arithmetic calculation\",\n", + " \"strict\": True, # Enable strict mode\n", + " \"parameters\": {\n", + " \"type\": \"object\",\n", + " \"properties\": {\n", + " \"operation\": {\"type\": \"string\", \"enum\": [\"add\", \"subtract\", \"multiply\", \"divide\"]},\n", + " \"a\": {\"type\": \"number\"},\n", + " \"b\": {\"type\": \"number\"},\n", + " },\n", + " \"required\": [\"operation\", \"a\", \"b\"],\n", + " },\n", + " }\n", + " ],\n", + "}\n", + "\n", + "# Create agents\n", + "calc_assistant = AssistantAgent(\n", + " name=\"MathAssistant\",\n", + " system_message=\"You solve math problems using tools and provide structured results.\",\n", + " llm_config=llm_config_combined,\n", + ")\n", + "\n", + "user_proxy_3 = UserProxyAgent(\n", + " name=\"User\",\n", + " human_input_mode=\"NEVER\",\n", + " max_consecutive_auto_reply=5,\n", + " code_execution_config=False,\n", + ")\n", + "\n", + "# Register function on both agents\n", + "calc_assistant.register_function({\"calculate\": calculate})\n", + "user_proxy_3.register_function({\"calculate\": calculate})\n", + "\n", + "print(\"✅ Example 3 configured: Combined strict tools + structured output\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "chat_result = user_proxy_3.run(\n", + " calc_assistant,\n", + " message=\"add 3 and 555\",\n", + " max_turns=2,\n", + ")\n", + "chat_result.process()" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Part 2: Vision and Image Input\n", + "\n", + "The V2 client also supports full multimodal capabilities with image input." + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Example 4: Simple Image Description\n", + "\n", + "Using formal image input format to reduce hallucination." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Configure LLM to use V2 client for vision\n", + "llm_config_vision = {\n", + " \"config_list\": [\n", + " {\n", + " \"api_type\": \"anthropic_v2\", # <-- Key: use V2 client architecture\n", + " \"model\": \"claude-3-5-haiku-20241022\", # Vision-capable model\n", + " \"api_key\": os.getenv(\"ANTHROPIC_API_KEY\"),\n", + " }\n", + " ],\n", + " \"temperature\": 0.3,\n", + "}\n", + "\n", + "# Create vision assistant\n", + "vision_assistant = AssistantAgent(\n", + " name=\"VisionBot\",\n", + " llm_config=llm_config_vision,\n", + " system_message=textwrap.dedent(\"\"\"\n", + " You are an AI assistant with vision capabilities.\n", + " You can analyze images and provide detailed, accurate descriptions.\n", + " \"\"\").strip(),\n", + ")\n", + "\n", + "# Create user proxy\n", + "user_proxy_vision = UserProxyAgent(\n", + " name=\"User\",\n", + " human_input_mode=\"NEVER\",\n", + " max_consecutive_auto_reply=0,\n", + " code_execution_config=False,\n", + ")\n", + "\n", + "# Test image URL\n", + "IMAGE_URL = \"https://upload.wikimedia.org/wikipedia/commons/3/3b/BlkStdSchnauzer2.jpg\"\n", + "\n", + "print(\"✅ Vision assistant with V2 client created\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Formal image input format (recommended)\n", + "message_with_image = {\n", + " \"role\": \"user\",\n", + " \"content\": [\n", + " {\"type\": \"text\", \"text\": \"Describe this image in one sentence.\"},\n", + " {\"type\": \"image_url\", \"image_url\": {\"url\": IMAGE_URL}},\n", + " ],\n", + "}\n", + "\n", + "# Initiate chat with image\n", + "chat_result = user_proxy_vision.initiate_chat(\n", + " vision_assistant, message=message_with_image, max_turns=1, summary_method=\"last_msg\"\n", + ")\n", + "\n", + "print(\"\\n=== Response ===\")\n", + "print(chat_result.summary)\n", + "print(f\"\\nCost: ${get_total_cost(chat_result.cost):.4f}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Example 5: Detailed Image Analysis" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "detailed_message = {\n", + " \"role\": \"user\",\n", + " \"content\": [\n", + " {\"type\": \"text\", \"text\": \"Analyze this image in detail. What breed is this dog? What are its characteristics?\"},\n", + " {\"type\": \"image_url\", \"image_url\": {\"url\": IMAGE_URL}},\n", + " ],\n", + "}\n", + "\n", + "chat_result = user_proxy_vision.initiate_chat(\n", + " vision_assistant,\n", + " message=detailed_message,\n", + " max_turns=1,\n", + " clear_history=True, # Start fresh conversation\n", + ")\n", + "\n", + "print(chat_result.summary)\n", + "print(f\"\\nCost: ${get_total_cost(chat_result.cost):.4f}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Summary\n", + "\n", + "### Key Benefits of Anthropic V2 Client\n", + "\n", + "1. **Structured Outputs**: Guaranteed schema-compliant JSON responses\n", + "2. **Strict Tool Use**: Type-safe function calls with guaranteed validation\n", + "3. **Vision Support**: Full multimodal capabilities with image input\n", + "4. **Rich Response Data**: Access to typed content blocks (reasoning, citations, etc.)\n", + "5. **Cost Tracking**: Automatic per-response cost calculation\n", + "6. **Type Safety**: Pydantic validation for all response data\n", + "7. **Forward Compatible**: GenericContent handles unknown future types\n", + "\n", + "### Usage Pattern\n", + "\n", + "on\n", + "# Simple: Just change api_type\n", + "```python\n", + "llm_config = {\n", + " \"config_list\": [{\n", + " \"api_type\": \"anthropic_v2\", # <-- That's it!\n", + " \"model\": \"claude-sonnet-4-5\",\n", + " \"api_key\": \"...\",\n", + " \"response_format\": YourPydanticModel, # Optional: for structured outputs\n", + " }]\n", + "}\n", + "\n", + "assistant = AssistantAgent(llm_config=llm_config)\n", + "# Everything works as before, but with rich UnifiedResponse internally\n", + "```\n", + "\n", + "### When to Use V2 Client\n", + "\n", + "- ✅ Need structured outputs with guaranteed schema compliance\n", + "- ✅ Want strict tool validation for type-safe function calls\n", + "- ✅ Working with vision/multimodal models\n", + "- ✅ Need access to reasoning blocks\n", + "- ✅ Require rich metadata and citations\n", + "- ✅ Building systems that need forward compatibility\n", + "\n", + "### Migration from Standard Client\n", + "\n", + "No code changes needed! Just update `api_type`:\n", + "```" + ] + } + ], + "metadata": { + "front_matter": { + "description": "Anthropic V2 Client.", + "tags": [ + "anthropic", + "ModelClientV2", + "client" + ] + }, + "kernelspec": { + "display_name": ".venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.14.2" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +} diff --git a/test/llm_clients/test_anthropic_v2.py b/test/llm_clients/test_anthropic_v2.py new file mode 100644 index 00000000000..fd97a711ce6 --- /dev/null +++ b/test/llm_clients/test_anthropic_v2.py @@ -0,0 +1,586 @@ +# test/llm_clients/test_anthropic_v2.py +# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors +# +# SPDX-License-Identifier: Apache-2.0 + +""" +Unit tests for AnthropicCompletionsClient (v2 client). + +These tests use mocked Anthropic API responses to test the client's +functionality without requiring API keys. +""" + +from typing import Any +from unittest.mock import Mock, patch + +import pytest +from pydantic import BaseModel + +from autogen.llm_clients.anthropic_v2 import AnthropicV2Client +from autogen.llm_clients.models import ( + GenericContent, + ReasoningContent, + TextContent, + ToolCallContent, + UnifiedMessage, + UnifiedResponse, +) + + +# Mock Anthropic response classes - use class names that match fallback checks +class TextBlock: + """Mock Anthropic TextBlock with correct class name.""" + + def __init__(self, text: str): + self.type = "text" + self.text = text + + +class ThinkingBlock: + """Mock Anthropic ThinkingBlock with correct class name.""" + + def __init__(self, thinking: str): + self.type = "thinking" + self.thinking = thinking + + +class ToolUseBlock: + """Mock Anthropic ToolUseBlock with correct class name.""" + + def __init__(self, tool_id: str, name: str, input_data: dict): + self.type = "tool_use" + self.id = tool_id + self.name = name + self.input = input_data + + +class MockUsage: + """Mock Anthropic usage object.""" + + def __init__(self, input_tokens: int = 10, output_tokens: int = 20): + self.input_tokens = input_tokens + self.output_tokens = output_tokens + + +class MockAnthropicMessage: + """Mock Anthropic Message response.""" + + def __init__( + self, + content: list[Any], + id: str = "msg_123", + model: str = "claude-3-5-sonnet-20241022", + stop_reason: str = "end_turn", + usage: MockUsage | None = None, + parsed_output: Any = None, + ): + self.id = id + self.model = model + self.content = content # Make sure this is a real list, not a Mock + self.stop_reason = stop_reason + self.usage = usage or MockUsage() + if parsed_output is not None: + self.parsed_output = parsed_output + + +@pytest.fixture +def mock_anthropic_client(): + """Create mock Anthropic client.""" + with patch("autogen.llm_clients.anthropic_v2.Anthropic") as mock_anthropic_class: + mock_client_instance = Mock() + mock_messages = Mock() + mock_beta = Mock() + mock_beta_messages = Mock() + mock_beta.messages = mock_beta_messages + mock_client_instance.messages = mock_messages + mock_client_instance.beta = mock_beta + mock_anthropic_class.return_value = mock_client_instance + yield mock_client_instance + + +@pytest.fixture(autouse=True) +def patch_helper_functions(): + """Patch helper functions to recognize mock block types.""" + + # Patch the helper functions to recognize our mock types + def _is_text_block_mock(content: Any) -> bool: + """Check if content is a TextBlock (mock or real).""" + # Check for our mock TextBlock first + if isinstance(content, TextBlock): + return True + # Fallback to original function for real types + from autogen.oai.anthropic import _is_text_block as original + + try: + return original(content) + except Exception: + return False + + def _is_thinking_block_mock(content: Any) -> bool: + """Check if content is a ThinkingBlock (mock or real).""" + # Check for our mock ThinkingBlock first + if isinstance(content, ThinkingBlock): + return True + # Fallback to original function for real types + from autogen.oai.anthropic import _is_thinking_block as original + + try: + return original(content) + except Exception: + return False + + def _is_tool_use_block_mock(content: Any) -> bool: + """Check if content is a ToolUseBlock (mock or real).""" + # Check for our mock ToolUseBlock first + if isinstance(content, ToolUseBlock): + return True + # Fallback to original function for real types + from autogen.oai.anthropic import _is_tool_use_block as original + + try: + return original(content) + except Exception: + return False + + with patch("autogen.llm_clients.anthropic_v2._is_text_block", side_effect=_is_text_block_mock): + with patch("autogen.llm_clients.anthropic_v2._is_thinking_block", side_effect=_is_thinking_block_mock): + with patch("autogen.llm_clients.anthropic_v2._is_tool_use_block", side_effect=_is_tool_use_block_mock): + yield + + +@pytest.fixture +def anthropic_v2_client(mock_anthropic_client): + """Create AnthropicV2Client instance with mocked Anthropic SDK.""" + return AnthropicV2Client(api_key="test-key") + + +class TestAnthropicV2ClientCreation: + """Test client initialization.""" + + def test_create_client_with_api_key(self, mock_anthropic_client): + """Test creating client with API key.""" + client = AnthropicV2Client(api_key="test-key") + assert client._api_key == "test-key" + assert client._client is not None + + def test_create_client_with_env_var(self, mock_anthropic_client): + """Test creating client with environment variable.""" + with patch.dict("os.environ", {"ANTHROPIC_API_KEY": "env-key"}): + client = AnthropicV2Client() + assert client._api_key == "env-key" + + def test_create_client_with_response_format(self, mock_anthropic_client): + """Test creating client with response format.""" + + class TestModel(BaseModel): + name: str + + client = AnthropicV2Client(api_key="test-key", response_format=TestModel) + assert client._response_format == TestModel + + def test_create_client_missing_api_key(self, mock_anthropic_client): + """Test creating client without API key raises error.""" + with patch.dict("os.environ", {}, clear=True), pytest.raises(ValueError, match="API key is required"): + AnthropicV2Client() + + +class TestStandardCompletion: + """Test standard completion without structured outputs.""" + + def test_simple_text_response(self, anthropic_v2_client, mock_anthropic_client): + """Test simple text response.""" + # Setup mock response + text_block = TextBlock("Hello, world!") + mock_response = MockAnthropicMessage(content=[text_block]) + mock_anthropic_client.messages.create.return_value = mock_response + + # Make request + params = { + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "Hello"}], + } + response = anthropic_v2_client.create(params) + + # Verify response + assert isinstance(response, UnifiedResponse) + assert response.model == "claude-3-5-sonnet-20241022" + assert response.provider == "anthropic" + assert len(response.messages) == 1 + assert len(response.messages[0].content) == 1 + assert isinstance(response.messages[0].content[0], TextContent) + assert response.messages[0].content[0].text == "Hello, world!" + + def test_response_with_thinking_block(self, anthropic_v2_client, mock_anthropic_client): + """Test response with thinking block.""" + # Setup mock response with thinking + thinking_block = ThinkingBlock("Let me think about this...") + text_block = TextBlock("The answer is 42") + mock_response = MockAnthropicMessage(content=[thinking_block, text_block]) + mock_anthropic_client.messages.create.return_value = mock_response + + # Make request + params = { + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "What is 6 * 7?"}], + } + response = anthropic_v2_client.create(params) + + # Verify response has both thinking and text + assert len(response.messages[0].content) == 2 + assert isinstance(response.messages[0].content[0], ReasoningContent) + assert response.messages[0].content[0].reasoning == "Let me think about this..." + assert isinstance(response.messages[0].content[1], TextContent) + assert response.messages[0].content[1].text == "The answer is 42" + + def test_response_with_tool_calls(self, anthropic_v2_client, mock_anthropic_client): + """Test response with tool calls.""" + # Setup mock response with tool use + tool_block = ToolUseBlock("tool_123", "get_weather", {"city": "San Francisco"}) + mock_response = MockAnthropicMessage(content=[tool_block], stop_reason="tool_use", usage=MockUsage(10, 15)) + mock_anthropic_client.messages.create.return_value = mock_response + + # Make request with tools + params = { + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "What's the weather in SF?"}], + "tools": [ + { + "name": "get_weather", + "description": "Get weather", + "input_schema": {"type": "object", "properties": {"city": {"type": "string"}}}, + } + ], + } + response = anthropic_v2_client.create(params) + + # Verify tool call + assert len(response.messages[0].content) == 1 + assert isinstance(response.messages[0].content[0], ToolCallContent) + assert response.messages[0].content[0].id == "tool_123" + assert response.messages[0].content[0].name == "get_weather" + assert response.finish_reason == "tool_calls" + + +class TestNativeStructuredOutputs: + """Test native structured outputs (beta API).""" + + def test_structured_output_with_parse_method(self, anthropic_v2_client, mock_anthropic_client): + """Test structured output using .parse() method (Pydantic model, no tools).""" + + class ContactInfo(BaseModel): + name: str + email: str + + # Setup mock parsed output + parsed_model = ContactInfo(name="John Doe", email="john@example.com") + mock_response = MockAnthropicMessage( + content=[TextBlock('{"name": "John Doe", "email": "john@example.com"}')], + parsed_output=parsed_model, + ) + mock_anthropic_client.beta.messages.parse.return_value = mock_response + + # Make request + params = { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "Extract contact info"}], + "response_format": ContactInfo, + } + response = anthropic_v2_client.create(params) + + # Verify parsed output is stored + assert len(response.messages[0].content) == 2 # GenericContent + TextContent + parsed_block = response.messages[0].content[0] + assert isinstance(parsed_block, GenericContent) + assert parsed_block.type == "parsed" + assert parsed_block.parsed == {"name": "John Doe", "email": "john@example.com"} + + def test_structured_output_with_create_method(self, anthropic_v2_client, mock_anthropic_client): + """Test structured output using .create() method (with tools or dict schema).""" + + class ContactInfo(BaseModel): + name: str + email: str + + # Setup mock response with JSON text (no parsed_output) + json_text = '{"name": "Jane Doe", "email": "jane@example.com"}' + mock_response = MockAnthropicMessage(content=[TextBlock(json_text)]) + # IMPORTANT: Set up .create() to return our mock response + mock_anthropic_client.beta.messages.create.return_value = mock_response + # Also set up .parse() in case it's called (though it shouldn't be with tools) + mock_anthropic_client.beta.messages.parse.return_value = mock_response + + # Make request with tools (forces .create() method) + params = { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "Extract contact info"}], + "response_format": ContactInfo, + "tools": [ + { + "name": "search", + "description": "Search", + "input_schema": {"type": "object", "properties": {}}, + } + ], + } + response = anthropic_v2_client.create(params) + + # Verify JSON was parsed into Pydantic model + assert len(response.messages[0].content) == 2 # GenericContent + TextContent + parsed_block = response.messages[0].content[0] + assert isinstance(parsed_block, GenericContent) + assert parsed_block.type == "parsed" + assert parsed_block.parsed == {"name": "Jane Doe", "email": "jane@example.com"} + + def test_structured_output_dict_schema(self, anthropic_v2_client, mock_anthropic_client): + """Test structured output with dict schema (always uses .create()).""" + # Setup mock response + json_text = '{"name": "Test", "value": 42}' + mock_response = MockAnthropicMessage(content=[TextBlock(json_text)]) + mock_anthropic_client.beta.messages.create.return_value = mock_response + # Also set up .parse() in case it's called (though it shouldn't be with dict schema) + mock_anthropic_client.beta.messages.parse.return_value = mock_response + + # Make request with dict schema + schema = { + "type": "object", + "properties": {"name": {"type": "string"}, "value": {"type": "integer"}}, + } + params = { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "Return JSON"}], + "response_format": schema, + } + response = anthropic_v2_client.create(params) + + # Verify text content (no parsing for dict schemas) + assert len(response.messages[0].content) == 1 + assert isinstance(response.messages[0].content[0], TextContent) + assert response.messages[0].content[0].text == json_text + + def test_structured_output_fallback_to_json_mode(self, anthropic_v2_client, mock_anthropic_client): + """Test fallback to JSON Mode when native SO fails.""" + + class TestModel(BaseModel): + value: str + + # Mock native SO to fail + from autogen.oai.anthropic import BadRequestError + + # Ensure both .create() and .parse() fail for native SO + mock_anthropic_client.beta.messages.create.side_effect = BadRequestError( + message="Error", body={}, response=Mock(status_code=400) + ) + mock_anthropic_client.beta.messages.parse.side_effect = BadRequestError( + message="Error", body={}, response=Mock(status_code=400) + ) + # Mock JSON Mode to succeed + json_mode_response = MockAnthropicMessage( + content=[TextBlock('{"value": "test"}')] + ) + mock_anthropic_client.messages.create.return_value = json_mode_response + + params = { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "Return JSON"}], + "response_format": TestModel, + } + response = anthropic_v2_client.create(params) + + # Should fallback to JSON Mode + assert mock_anthropic_client.messages.create.called + + +class TestJSONMode: + """Test JSON Mode structured outputs (fallback for older models).""" + + def test_json_mode_extraction(self, anthropic_v2_client, mock_anthropic_client): + """Test JSON extraction from tags.""" + + class TestModel(BaseModel): + name: str + age: int + + # Setup mock response with JSON in tags + json_text = '{"name": "Alice", "age": 30}' + mock_response = MockAnthropicMessage(content=[TextBlock(json_text)]) + mock_anthropic_client.messages.create.return_value = mock_response + + # Patch supports_native_structured_outputs to return False for older model + with patch("autogen.llm_clients.anthropic_v2.supports_native_structured_outputs", return_value=False): + params = { + "model": "claude-3-5-sonnet-20241022", # Older model + "messages": [{"role": "user", "content": "Return JSON"}], + "response_format": TestModel, + } + response = anthropic_v2_client.create(params) + + # Verify JSON was extracted and parsed + text_content = response.messages[0].content[0] + assert isinstance(text_content, TextContent) + # Should contain parsed JSON (not the tags) + assert "name" in text_content.text + assert "Alice" in text_content.text + + +class TestCostAndUsage: + """Test cost calculation and usage extraction.""" + + def test_cost_calculation(self, anthropic_v2_client): + """Test cost calculation from usage.""" + usage = {"prompt_tokens": 100, "completion_tokens": 200, "total_tokens": 300} + response = UnifiedResponse( + id="test", + model="claude-3-5-sonnet-20241022", + provider="anthropic", + messages=[UnifiedMessage(role="assistant", content=[TextContent(type="text", text="test")])], + usage=usage, + ) + cost = anthropic_v2_client.cost(response) + assert isinstance(cost, float) + assert cost >= 0 + + def test_get_usage(self, anthropic_v2_client): + """Test get_usage static method.""" + usage = {"prompt_tokens": 50, "completion_tokens": 75, "total_tokens": 125} + response = UnifiedResponse( + id="test", + model="claude-3-5-sonnet-20241022", + provider="anthropic", + messages=[UnifiedMessage(role="assistant", content=[TextContent(type="text", text="test")])], + usage=usage, + cost=0.001, + ) + usage_dict = AnthropicV2Client.get_usage(response) + assert usage_dict["prompt_tokens"] == 50 + assert usage_dict["completion_tokens"] == 75 + assert usage_dict["total_tokens"] == 125 + assert usage_dict["cost"] == 0.001 + assert usage_dict["model"] == "claude-3-5-sonnet-20241022" + + +class TestMessageRetrieval: + """Test message_retrieval method.""" + + def test_message_retrieval_text_only(self, anthropic_v2_client): + """Test message retrieval for text-only response.""" + response = UnifiedResponse( + id="test", + model="claude-3-5-sonnet-20241022", + provider="anthropic", + messages=[ + UnifiedMessage( + role="assistant", + content=[TextContent(type="text", text="Hello, world!")], + ) + ], + usage={"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, + ) + messages = anthropic_v2_client.message_retrieval(response) + assert isinstance(messages, list) + assert len(messages) == 1 + assert messages[0] == "Hello, world!" + + def test_message_retrieval_with_tool_calls(self, anthropic_v2_client): + """Test message retrieval with tool calls.""" + response = UnifiedResponse( + id="test", + model="claude-3-5-sonnet-20241022", + provider="anthropic", + messages=[ + UnifiedMessage( + role="assistant", + content=[ + ToolCallContent( + type="tool_call", + id="call_123", + name="get_weather", + arguments='{"city": "SF"}', + ) + ], + ) + ], + usage={"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30}, + ) + messages = anthropic_v2_client.message_retrieval(response) + assert isinstance(messages, list) + assert len(messages) == 1 + # Should return ChatCompletionMessage dict for tool calls + assert hasattr(messages[0], "tool_calls") or isinstance(messages[0], dict) + + +class TestV1Compatibility: + """Test create_v1_compatible method.""" + + def test_create_v1_compatible(self, anthropic_v2_client, mock_anthropic_client): + """Test create_v1_compatible returns ChatCompletion format.""" + # Setup mock response + text_block = TextBlock("Test response") + mock_response = MockAnthropicMessage(content=[text_block]) + mock_anthropic_client.messages.create.return_value = mock_response + + params = { + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "Hello"}], + } + v1_response = anthropic_v2_client.create_v1_compatible(params) + + # Verify ChatCompletion format + from autogen.oai.oai_models import ChatCompletion + + assert isinstance(v1_response, ChatCompletion) + assert v1_response.id is not None + assert v1_response.model == "claude-3-5-sonnet-20241022" + assert len(v1_response.choices) == 1 + assert v1_response.choices[0].message.content == "Test response" + + def test_create_v1_compatible_with_thinking(self, anthropic_v2_client, mock_anthropic_client): + """Test create_v1_compatible preserves thinking blocks.""" + # Setup mock response with thinking + thinking_block = ThinkingBlock("Thinking...") + text_block = TextBlock("Answer") + mock_response = MockAnthropicMessage(content=[thinking_block, text_block]) + mock_anthropic_client.messages.create.return_value = mock_response + + params = { + "model": "claude-3-5-sonnet-20241022", + "messages": [{"role": "user", "content": "Question"}], + } + v1_response = anthropic_v2_client.create_v1_compatible(params) + + # Verify thinking is preserved in [Thinking] tags + content = v1_response.choices[0].message.content + assert "[Thinking]" in content + assert "Thinking..." in content + assert "Answer" in content + + +class TestErrorHandling: + """Test error handling.""" + + def test_invalid_json_in_structured_output(self, anthropic_v2_client, mock_anthropic_client): + """Test handling of invalid JSON in structured output.""" + + class TestModel(BaseModel): + value: str + + # Setup mock response with invalid JSON + mock_response = MockAnthropicMessage(content=[TextBlock("not valid json")]) + mock_anthropic_client.beta.messages.create.return_value = mock_response + # Also set up .parse() in case it's called (though it shouldn't be with tools) + mock_anthropic_client.beta.messages.parse.return_value = mock_response + + params = { + "model": "claude-sonnet-4-5", + "messages": [{"role": "user", "content": "Return JSON"}], + "response_format": TestModel, + "tools": [{"name": "test", "input_schema": {}}], # Force .create() method + } + # Should not raise, but log warning + response = anthropic_v2_client.create(params) + # Should fallback to raw text + assert isinstance(response.messages[0].content[0], TextContent) + + def test_missing_api_key_error(self): + """Test error when API key is missing.""" + with patch.dict("os.environ", {}, clear=True), pytest.raises(ValueError, match="API key is required"): + AnthropicV2Client() diff --git a/test/llm_clients/test_anthropic_v2_integration.py b/test/llm_clients/test_anthropic_v2_integration.py new file mode 100644 index 00000000000..d2fa8fc855a --- /dev/null +++ b/test/llm_clients/test_anthropic_v2_integration.py @@ -0,0 +1,494 @@ +# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors +# +# SPDX-License-Identifier: Apache-2.0 + +""" +Integration tests for AnthropicCompletionsClient (V2) with real API calls. + +These tests require: +- ANTHROPIC_API_KEY environment variable set +- Anthropic account with access to Claude Sonnet 4.5+ models +- pytest markers: @pytest.mark.anthropic, @pytest.mark.integration +- @run_for_optional_imports decorator to handle optional dependencies + +Run with: + pytest test/llm_clients/test_anthropic_v2_integration.py -m "anthropic and integration" +""" + +import json + +import pytest +from pydantic import BaseModel + +from autogen import AssistantAgent, UserProxyAgent +from autogen.import_utils import run_for_optional_imports +from test.credentials import Credentials + + +@pytest.fixture +def anthropic_v2_llm_config(credentials_anthropic_claude_sonnet: Credentials) -> dict: + """Create LLM config for Anthropic V2 client.""" + # Skip if credentials are not available or invalid + try: + api_key = credentials_anthropic_claude_sonnet.api_key + if not api_key or api_key == "": + pytest.skip("ANTHROPIC_API_KEY not set and OAI_CONFIG_LIST file not found") + except (AttributeError, FileNotFoundError, Exception) as e: + pytest.skip(f"Could not load Anthropic credentials: {e}") + + return { + "config_list": [ + { + "model": "claude-sonnet-4-5", + "api_key": credentials_anthropic_claude_sonnet.api_key, + "api_type": "anthropic_v2", + } + ], + } + + +@pytest.fixture +def anthropic_v2_llm_config_vision(credentials_anthropic_claude_sonnet: Credentials) -> dict: + """Create LLM config for Anthropic V2 client with vision model.""" + # Skip if credentials are not available or invalid + try: + api_key = credentials_anthropic_claude_sonnet.api_key + if not api_key or api_key == "": + pytest.skip("ANTHROPIC_API_KEY not set and OAI_CONFIG_LIST file not found") + except (AttributeError, FileNotFoundError, Exception) as e: + pytest.skip(f"Could not load Anthropic credentials: {e}") + + return { + "config_list": [ + { + "api_type": "anthropic_v2", + "model": "claude-3-5-haiku-20241022", # Vision-capable model + "api_key": credentials_anthropic_claude_sonnet.api_key, + } + ], + "temperature": 0.3, + } + + +class TestAnthropicV2StructuredOutputs: + """Test structured outputs with Pydantic models (from notebook Example 1).""" + + @pytest.mark.anthropic + @pytest.mark.integration + @run_for_optional_imports("anthropic", "anthropic") + def test_structured_output_math_reasoning(self, anthropic_v2_llm_config): + """Test structured output with math reasoning using Pydantic models.""" + + # Define the structured output schema + class Step(BaseModel): + """A single step in mathematical reasoning.""" + + explanation: str + output: str + + class MathReasoning(BaseModel): + """Structured output for mathematical problem solving.""" + + steps: list[Step] + final_answer: str + + def format(self) -> str: + """Format the response for display.""" + steps_output = "\n".join( + f"Step {i + 1}: {step.explanation}\n Output: {step.output}" for i, step in enumerate(self.steps) + ) + return f"{steps_output}\n\nFinal Answer: {self.final_answer}" + + # Configure LLM with structured output + llm_config = anthropic_v2_llm_config.copy() + llm_config["config_list"][0]["response_format"] = MathReasoning + + # Create agents + user_proxy = UserProxyAgent( + name="User", + human_input_mode="NEVER", + max_consecutive_auto_reply=0, + code_execution_config=False, + ) + + math_assistant = AssistantAgent( + name="MathAssistant", + system_message="You are a math tutor. Solve problems step by step.", + llm_config=llm_config, + ) + + # Ask the assistant to solve a math problem + chat_result = user_proxy.run( + math_assistant, + message="Solve the equation: 3x + 7 = 22", + max_turns=1, + ) + + # Process the response to populate messages, summary, and cost + chat_result.process() + + # Verify chat result + assert chat_result is not None + assert chat_result.messages is not None + messages_list = list(chat_result.messages) + assert len(messages_list) > 0 + + # Verify the response contains structured output + # The response should be formatted by the MathReasoning.format() method + last_message = messages_list[-1] + assert "content" in last_message or "text" in str(last_message) + + # Verify cost tracking + assert chat_result.cost is not None + assert chat_result.cost.usage_including_cached_inference.total_cost >= 0 + + +class TestAnthropicV2StrictToolUse: + """Test strict tool use for type-safe function calls (from notebook Example 2).""" + + @pytest.mark.anthropic + @pytest.mark.integration + @run_for_optional_imports("anthropic", "anthropic") + def test_strict_tool_use_weather(self, anthropic_v2_llm_config): + """Test strict tool use with weather API.""" + + # Define a tool function + def get_weather(location: str, unit: str = "celsius") -> str: + """Get the weather for a location. + + Args: + location: The city and state, e.g. San Francisco, CA + unit: Temperature unit (celsius or fahrenheit) + """ + return f"Weather in {location}: 22°{unit.upper()[0]}, partly cloudy" + + # Configure LLM with strict tool + llm_config_strict = anthropic_v2_llm_config.copy() + llm_config_strict["functions"] = [ + { + "name": "get_weather", + "description": "Get the weather for a location", + "strict": True, # Enable strict schema validation + "parameters": { + "type": "object", + "properties": { + "location": {"type": "string", "description": "The city and state, e.g. San Francisco, CA"}, + "unit": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "Temperature unit", + }, + }, + "required": ["location"], + }, + } + ] + + # Create agents + weather_assistant = AssistantAgent( + name="WeatherAssistant", + system_message="You help users get weather information. Use the get_weather function.", + llm_config=llm_config_strict, + ) + + user_proxy = UserProxyAgent( + name="User", + human_input_mode="NEVER", + max_consecutive_auto_reply=1, + code_execution_config=False, + ) + + # Register function on both agents + weather_assistant.register_function({"get_weather": get_weather}) + user_proxy.register_function({"get_weather": get_weather}) + + # Query the weather + chat_result = user_proxy.initiate_chat( + weather_assistant, + message="What's the weather in Boston, MA?", + max_turns=2, + ) + + # Verify chat result + assert chat_result is not None + assert chat_result.chat_history is not None + assert len(chat_result.chat_history) > 0 + + # Verify tool call was made with correct types + tool_call_found = False + for message in chat_result.chat_history: + if message.get("tool_calls"): + tool_call = message["tool_calls"][0] + args = json.loads(tool_call["function"]["arguments"]) + assert tool_call["function"]["name"] == "get_weather" + assert "location" in args + assert isinstance(args["location"], str) + assert args["location"] == "Boston, MA" or "Boston" in args["location"] + # If unit is provided, it should be a valid enum value + if "unit" in args: + assert args["unit"] in ["celsius", "fahrenheit"] + tool_call_found = True + break + + assert tool_call_found, "Tool call should have been made" + + # Verify cost tracking + assert chat_result.cost is not None + total_cost = sum( + model_usage.get("cost", 0) + for usage_type in chat_result.cost.values() + if isinstance(usage_type, dict) + for model_usage in usage_type.values() + if isinstance(model_usage, dict) + ) + assert total_cost >= 0 + + +class TestAnthropicV2CombinedFeatures: + """Test combined structured outputs + strict tools (from notebook Example 3).""" + + @pytest.mark.anthropic + @pytest.mark.integration + @run_for_optional_imports("anthropic", "anthropic") + def test_combined_structured_output_and_strict_tools(self, anthropic_v2_llm_config): + """Test combined strict tools + structured output.""" + + # Define calculator tool + def calculate(operation: str, a: float, b: float) -> float: + """Perform a calculation. + + Args: + operation: The operation to perform (add, subtract, multiply, divide) + a: First number + b: Second number + """ + if operation == "add": + return a + b + elif operation == "subtract": + return a - b + elif operation == "multiply": + return a * b + elif operation == "divide": + return a / b if b != 0 else 0 + return 0 + + # Result model for structured output + class CalculationResult(BaseModel): + """Structured output for calculation results.""" + + problem: str + steps: list[str] + result: float + verification: str + + # Configure with BOTH features + llm_config_combined = anthropic_v2_llm_config.copy() + llm_config_combined["config_list"][0]["response_format"] = CalculationResult + llm_config_combined["functions"] = [ + { + "name": "calculate", + "description": "Perform arithmetic calculation", + "strict": True, # Enable strict mode + "parameters": { + "type": "object", + "properties": { + "operation": {"type": "string", "enum": ["add", "subtract", "multiply", "divide"]}, + "a": {"type": "number"}, + "b": {"type": "number"}, + }, + "required": ["operation", "a", "b"], + }, + } + ] + + # Create agents + calc_assistant = AssistantAgent( + name="MathAssistant", + system_message="You solve math problems using tools and provide structured results.", + llm_config=llm_config_combined, + ) + + user_proxy = UserProxyAgent( + name="User", + human_input_mode="NEVER", + max_consecutive_auto_reply=5, + code_execution_config=False, + ) + + # Register function on both agents + calc_assistant.register_function({"calculate": calculate}) + user_proxy.register_function({"calculate": calculate}) + + # Run calculation + chat_result = user_proxy.run( + calc_assistant, + message="add 3 and 555", + max_turns=2, + ) + + # Process the response to populate messages, summary, and cost + chat_result.process() + + # Verify chat result + assert chat_result is not None + assert chat_result.messages is not None + messages_list = list(chat_result.messages) + assert len(messages_list) > 0 + + # Verify tool call was made + tool_call_found = False + for message in messages_list: + if message.get("tool_calls"): + tool_call = message["tool_calls"][0] + args = json.loads(tool_call["function"]["arguments"]) + assert tool_call["function"]["name"] == "calculate" + assert args["operation"] == "add" + assert isinstance(args["a"], (int, float)) + assert isinstance(args["b"], (int, float)) + tool_call_found = True + break + + assert tool_call_found, "Tool call should have been made" + + # Verify cost tracking + assert chat_result.cost is not None + assert chat_result.cost.usage_including_cached_inference.total_cost >= 0 + + +class TestAnthropicV2Vision: + """Test vision and image input capabilities (from notebook Examples 4-5).""" + + @pytest.mark.anthropic + @pytest.mark.integration + @run_for_optional_imports("anthropic", "anthropic") + def test_vision_simple_image_description(self, anthropic_v2_llm_config_vision): + """Test simple image description with vision model.""" + # Create vision assistant + vision_assistant = AssistantAgent( + name="VisionBot", + llm_config=anthropic_v2_llm_config_vision, + system_message="You are an AI assistant with vision capabilities. You can analyze images and provide detailed, accurate descriptions.", + ) + + # Create user proxy + user_proxy_vision = UserProxyAgent( + name="User", + human_input_mode="NEVER", + max_consecutive_auto_reply=0, + code_execution_config=False, + ) + + # Test image URL + IMAGE_URL = "https://upload.wikimedia.org/wikipedia/commons/3/3b/BlkStdSchnauzer2.jpg" + + # Formal image input format + message_with_image = { + "role": "user", + "content": [ + {"type": "text", "text": "Describe this image in one sentence."}, + {"type": "image_url", "image_url": {"url": IMAGE_URL}}, + ], + } + + # Initiate chat with image + chat_result = user_proxy_vision.initiate_chat( + vision_assistant, message=message_with_image, max_turns=1, summary_method="last_msg" + ) + + # Verify chat result + assert chat_result is not None + assert chat_result.summary is not None + assert len(chat_result.summary) > 0 + + # Verify the response mentions something about the image (dog, schnauzer, etc.) + summary_lower = chat_result.summary.lower() + assert any( + keyword in summary_lower for keyword in ["dog", "schnauzer", "animal", "pet", "image", "photo", "picture"] + ) + + # Verify cost tracking + assert chat_result.cost is not None + total_cost = sum( + model_usage.get("cost", 0) + for usage_type in chat_result.cost.values() + if isinstance(usage_type, dict) + for model_usage in usage_type.values() + if isinstance(model_usage, dict) + ) + assert total_cost >= 0 + + @pytest.mark.anthropic + @pytest.mark.integration + @run_for_optional_imports("anthropic", "anthropic") + def test_vision_detailed_image_analysis(self, anthropic_v2_llm_config_vision): + """Test detailed image analysis.""" + # Create vision assistant + vision_assistant = AssistantAgent( + name="VisionBot", + llm_config=anthropic_v2_llm_config_vision, + system_message="You are an AI assistant with vision capabilities. You can analyze images and provide detailed, accurate descriptions.", + ) + + # Create user proxy + user_proxy_vision = UserProxyAgent( + name="User", + human_input_mode="NEVER", + max_consecutive_auto_reply=0, + code_execution_config=False, + ) + + # Test image URL + IMAGE_URL = "https://upload.wikimedia.org/wikipedia/commons/3/3b/BlkStdSchnauzer2.jpg" + + # Detailed analysis request + detailed_message = { + "role": "user", + "content": [ + { + "type": "text", + "text": "Analyze this image in detail. What breed is this dog? What are its characteristics?", + }, + {"type": "image_url", "image_url": {"url": IMAGE_URL}}, + ], + } + + # Initiate chat + chat_result = user_proxy_vision.initiate_chat( + vision_assistant, + message=detailed_message, + max_turns=1, + clear_history=True, # Start fresh conversation + ) + + # Verify chat result + assert chat_result is not None + assert chat_result.summary is not None + assert len(chat_result.summary) > 0 + + # Verify the response contains detailed analysis + summary_lower = chat_result.summary.lower() + # Should mention breed or characteristics + assert any( + keyword in summary_lower + for keyword in [ + "schnauzer", + "breed", + "dog", + "characteristic", + "feature", + "appearance", + "black", + "standard", + ] + ) + + # Verify cost tracking + assert chat_result.cost is not None + total_cost = sum( + model_usage.get("cost", 0) + for usage_type in chat_result.cost.values() + if isinstance(usage_type, dict) + for model_usage in usage_type.values() + if isinstance(model_usage, dict) + ) + assert total_cost >= 0 diff --git a/website/docs/user-guide/basic-concepts/structured-outputs.mdx b/website/docs/user-guide/basic-concepts/structured-outputs.mdx index 0dad0b3bfd0..2294c7c22d3 100644 --- a/website/docs/user-guide/basic-concepts/structured-outputs.mdx +++ b/website/docs/user-guide/basic-concepts/structured-outputs.mdx @@ -37,7 +37,13 @@ Just as standardized medical forms ensure complete and consistent documentation Implementing structured outputs in AG2 is straightforward using Pydantic models and the [`response_format`](/docs/api-reference/autogen/llm_config/LLMConfigEntry#response_format) parameter: -Not all model providers support structured outputs. For a current list of supported providers, see [here](https://docs.ag2.ai/latest/docs/use-cases/notebooks/notebooks/agentchat_structured_outputs/?h=structured#supported-model-providers). +Not all model providers support structured outputs. Supported providers include: + +- **OpenAI**: Native structured outputs via `response_format` parameter +- **Anthropic V2** (`api_type: "anthropic_v2"`): Native structured outputs using Anthropic's beta API with Pydantic models or JSON schemas +- **Anthropic V1** (`api_type: "anthropic"`): JSON Mode (prompt-based fallback) + +For a current list of supported providers and examples, see [here](https://docs.ag2.ai/latest/docs/use-cases/notebooks/notebooks/agentchat_structured_outputs/?h=structured#supported-model-providers). ```python hl_lines="5-8 14" @@ -600,6 +606,43 @@ Structured outputs offer several compelling benefits: - **Integration**: Easy to parse and use in downstream systems - **Maintainability**: Schema changes can be managed centrally +## Provider-Specific Features + +### Anthropic V2 Native Structured Outputs + +When using `api_type: "anthropic_v2"`, structured outputs are handled natively by Anthropic's beta API. This provides several advantages: + +- **Guaranteed Schema Compliance**: The API enforces schema validation at the API level, not just through prompts +- **No Retries Needed**: Invalid responses are prevented, not just detected +- **Dual Methods**: Supports both `.parse()` (for Pydantic models) and `.create()` (for JSON schemas or when tools are present) +- **Strict Tool Use**: Can combine structured outputs with strict tool validation + +Example with Anthropic V2: + +```python +from pydantic import BaseModel +from autogen import LLMConfig, AssistantAgent +import os + +class ContactInfo(BaseModel): + name: str + email: str + plan_interest: str + +llm_config = LLMConfig( + config_list={ + "model": "claude-sonnet-4-5", + "api_type": "anthropic_v2", # Use V2 client for native support + "api_key": os.getenv("ANTHROPIC_API_KEY"), + }, + response_format=ContactInfo, +) + +agent = AssistantAgent("assistant", llm_config=llm_config) +``` + +For more details on Anthropic V2 structured outputs, see the [Anthropic V2 Client documentation](/docs/user-guide/models/anthropic#anthropic-v2-client). + ## Next Steps You've now learned how to create a complete financial compliance system incorporating the basic concepts of AG2: diff --git a/website/docs/user-guide/models/anthropic.mdx b/website/docs/user-guide/models/anthropic.mdx index 53b927e1ed9..1d3697a7ad7 100644 --- a/website/docs/user-guide/models/anthropic.mdx +++ b/website/docs/user-guide/models/anthropic.mdx @@ -862,3 +862,281 @@ response.process() # Print out the final response (thinking tokens are not shown) print(response.summary) ``` + +## Anthropic V2 Client + +The Anthropic V2 client (`api_type: "anthropic_v2"`) provides enhanced features and a modern API interface for working with Claude models. It implements the `ModelClientV2` protocol, returning rich `UnifiedResponse` objects with typed content blocks. + +### Features + +- **Native Structured Outputs**: Guaranteed schema-compliant JSON responses using Pydantic models or JSON schemas +- **Strict Tool Use**: Type-safe function calls with guaranteed schema validation +- **Vision Support**: Full multimodal capabilities with image input +- **Rich Response Objects**: Typed content blocks including `TextContent`, `ReasoningContent`, `ToolCallContent`, and `CitationContent` +- **Extended Thinking**: Support for Claude's thinking mode with reasoning blocks +- **Cost Tracking**: Automatic per-response cost calculation + +### Requirements + +To use the Anthropic V2 client, you need: + +- `anthropic>=0.74.1` (for structured outputs support) +- Claude Sonnet 4.5 (`claude-sonnet-4-5`) or Claude Opus 4.1 (`claude-opus-4-1`) for structured outputs +- Set `api_type: "anthropic_v2"` in your configuration + +### Basic Configuration + +```python +import os +from autogen import LLMConfig + +llm_config = LLMConfig(config_list={ + "model": "claude-sonnet-4-5", + "api_key": os.getenv("ANTHROPIC_API_KEY"), + "api_type": "anthropic_v2", # Use V2 client +}) +``` + +### Structured Outputs with Pydantic Models + +The V2 client supports native structured outputs using Pydantic models. This ensures guaranteed schema compliance and type safety: + +```python +from pydantic import BaseModel +from autogen import AssistantAgent, UserProxyAgent, LLMConfig + +# Define structured output schema +class Step(BaseModel): + explanation: str + output: str + +class MathReasoning(BaseModel): + steps: list[Step] + final_answer: str + +# Configure with structured output +llm_config = LLMConfig( + config_list={ + "model": "claude-sonnet-4-5", + "api_key": os.getenv("ANTHROPIC_API_KEY"), + "api_type": "anthropic_v2", + }, + response_format=MathReasoning, # Enable structured outputs +) + +math_assistant = AssistantAgent( + name="MathAssistant", + system_message="You are a math tutor. Solve problems step by step.", + llm_config=llm_config, +) + +user_proxy = UserProxyAgent( + name="User", + human_input_mode="NEVER", + max_consecutive_auto_reply=0, +) + +# Use the agent - responses will match MathReasoning schema +chat_result = user_proxy.run( + math_assistant, + message="Solve the equation: 3x + 7 = 22", + max_turns=1, +) +chat_result.process() +print(chat_result.summary) +``` + +### Strict Tool Use + +The V2 client supports strict tool use, ensuring that tool inputs exactly match your schema: + +```python +from typing import Annotated +from autogen import AssistantAgent, UserProxyAgent, LLMConfig + +# Configure with V2 client +llm_config = LLMConfig(config_list={ + "model": "claude-sonnet-4-5", + "api_key": os.getenv("ANTHROPIC_API_KEY"), + "api_type": "anthropic_v2", +}) + +assistant = AssistantAgent("assistant", llm_config=llm_config) +user_proxy = UserProxyAgent("user_proxy", human_input_mode="NEVER") + +# Define tool with strict validation +@user_proxy.register_for_execution() +@assistant.register_for_llm( + name="get_weather", + description="Get the current weather in a given location.", + strict=True, # Enable strict tool use +) +def get_weather(location: Annotated[str, "The city and state, e.g. Toronto, ON."]) -> str: + return f"Weather in {location}: Sunny, 22°C" + +user_proxy.initiate_chat( + assistant, + message="What's the weather in Toronto?", +) +``` + +### Vision Support + +The V2 client supports image input for vision-capable models: + +```python +from autogen import AssistantAgent, UserProxyAgent, LLMConfig + +llm_config = LLMConfig(config_list={ + "model": "claude-3-5-haiku-20241022", # Vision-capable model + "api_key": os.getenv("ANTHROPIC_API_KEY"), + "api_type": "anthropic_v2", +}) + +vision_assistant = AssistantAgent( + name="VisionAssistant", + system_message="You are a helpful assistant that can analyze images.", + llm_config=llm_config, +) + +user_proxy = UserProxyAgent("user_proxy", human_input_mode="NEVER") + +# Message with image +message_with_image = { + "role": "user", + "content": [ + { + "type": "text", + "text": "What's in this image?", + }, + { + "type": "image_url", + "image_url": { + "url": "https://example.com/image.jpg", # Or use base64 data URI + }, + }, + ], +} + +chat_result = user_proxy.initiate_chat( + vision_assistant, + message=message_with_image, + max_turns=1, +) +``` + +### Differences from V1 Client + +| Feature | V1 Client (`anthropic`) | V2 Client (`anthropic_v2`) | +|---------|------------------------|----------------------------| +| Response Format | OpenAI-compatible `ChatCompletion` | Rich `UnifiedResponse` with typed blocks | +| Structured Outputs | JSON Mode (prompt-based) | Native structured outputs (beta API) | +| Tool Validation | Basic validation | Strict tool use with guaranteed schema compliance | +| Content Blocks | Text only | Text, Reasoning, Tool Calls, Citations | +| Vision Support | Limited | Full multimodal support | +| Cost Tracking | Per-request | Per-response with detailed breakdown | + +### When to Use V1 vs V2 + +AG2 now offers an Anthropic V2 client that returns richer response objects. Here's when to use each: + +**When to use V2:** + +- When you need access to reasoning blocks separately +- When building custom agents that process structured content +- For forward compatibility with future Anthropic features +- When you need native structured outputs with guaranteed schema compliance +- When you require strict tool use with type-safe function calls +- For vision capabilities and multimodal interactions + +**When to use V1:** + +- When you need AWS Bedrock or GCP Vertex AI support (V2 support coming soon) +- When your existing code expects `ChatCompletion` format +- For simple use cases where `UnifiedResponse` is overkill +- When working with legacy Anthropic API features +- For basic text generation without structured outputs + +### Benefits of UnifiedResponse over ChatCompletion + +The V2 client's `UnifiedResponse` provides several advantages over the V1 client's `ChatCompletion` format: + +1. **Typed Content Blocks**: Access to structured content types including: + - `TextContent`: Standard text responses + - `ReasoningContent`: Extended thinking/reasoning blocks (when using thinking mode) + - `ToolCallContent`: Structured tool/function calls + - `CitationContent`: Citations and references + +2. **Better Type Safety**: Typed response objects enable better IDE support and catch errors at development time + +3. **Richer Metadata**: More detailed information about the response, including per-block metadata + +4. **Future-Proof**: Designed to support new Anthropic API features as they're released + +5. **Structured Outputs**: Native support for guaranteed schema-compliant JSON responses without prompt engineering + +### Example: Combined Structured Outputs and Tools + +You can combine structured outputs with strict tool use for powerful agentic workflows: + +```python +from pydantic import BaseModel +from typing import Annotated +from autogen import AssistantAgent, UserProxyAgent, LLMConfig + +# Structured output for calculation results +class CalculationResult(BaseModel): + operation: str + operands: list[float] + result: float + +# Configure with both structured outputs and tools +llm_config = LLMConfig( + config_list={ + "model": "claude-sonnet-4-5", + "api_key": os.getenv("ANTHROPIC_API_KEY"), + "api_type": "anthropic_v2", + }, + response_format=CalculationResult, +) + +calc_assistant = AssistantAgent( + name="Calculator", + system_message="You are a calculator assistant. Use tools to perform calculations.", + llm_config=llm_config, +) + +user_proxy = UserProxyAgent("user_proxy", human_input_mode="NEVER") + + +# Tool for calculations +@user_proxy.register_for_execution() +@calc_assistant.register_for_llm( + name="calculate", + description="Perform arithmetic operations", + strict=True, +) +def calculate( + operation: Annotated[str, "Operation: add, subtract, multiply, divide"], + a: Annotated[float, "First number"], + b: Annotated[float, "Second number"], +) -> float: + if operation == "add": + return a + b + elif operation == "subtract": + return a - b + elif operation == "multiply": + return a * b + elif operation == "divide": + return a / b if b != 0 else float("inf") + return 0.0 + +chat_result = user_proxy.run( + calc_assistant, + message="add 3 and 555", + max_turns=2, +) +chat_result.process() +``` + +For more examples, see the [Anthropic V2 Client notebook](/docs/use-cases/notebooks/notebooks/agentchat_anthropic_v2_client_example).