Anthropic

Anthropic

Bases: BaseLLM

Anthropic LLM node.

This class provides an implementation for the Anthropic Language Model node.

Attributes:

Name	Type	Description
connection	Anthropic | None	The connection to use for the Anthropic LLM.
cache_control	AnthropicCacheControl | None	The cache control configuration.
strict_tools	bool | list[str]	Inherited from :class:BaseLLM. False (default, or an empty list) ships every tool as-is with no strict guarantee; True cleans each tool's schema to Anthropic's strict subset and attaches strict: true (up to :data:ANTHROPIC_MAX_STRICT_TOOLS per request); a list of tool (function) names makes only those tools strict and ships the rest untouched. Use a list to exclude tools whose schema exceeds Anthropic's strict grammar-compilation budget (the Schema is too complex for compilation 400).

Source code in dynamiq/nodes/llms/anthropic.py

class Anthropic(BaseLLM):
    """Anthropic LLM node.

    This class provides an implementation for the Anthropic Language Model node.

    Attributes:
        connection (AnthropicConnection | None): The connection to use for the Anthropic LLM.
        cache_control (AnthropicCacheControl | None): The cache control configuration.
        strict_tools: Inherited from :class:`BaseLLM`. False (default, or an empty
            list) ships every tool as-is with no strict guarantee; True cleans each
            tool's schema to Anthropic's strict subset and attaches ``strict: true``
            (up to :data:`ANTHROPIC_MAX_STRICT_TOOLS` per request); a list of tool
            (function) names makes only those tools strict and ships the rest
            untouched. Use a list to exclude tools whose schema exceeds Anthropic's
            strict grammar-compilation budget (the ``Schema is too complex for
            compilation`` 400).
    """

    connection: AnthropicConnection | None = None
    MODEL_PREFIX = "anthropic/"
    MAX_STRICT_TOOLS: ClassVar[int] = ANTHROPIC_MAX_STRICT_TOOLS
    cache_control: AnthropicCacheControl | None = None

    def __init__(self, **kwargs):
        """Initialize the Anthropic LLM node.

        Args:
            **kwargs: Additional keyword arguments.
        """
        if kwargs.get("client") is None and kwargs.get("connection") is None:
            kwargs["connection"] = AnthropicConnection()
        super().__init__(**kwargs)

    @staticmethod
    def _convert_non_image_to_file_content(messages: list[dict]) -> list[dict]:
        for message in messages:
            content = message.get("content")
            if not isinstance(content, list):
                continue

            new_content = []
            for item in content:
                if isinstance(item, dict) and item.get("type") == VisionMessageType.IMAGE_URL and "image_url" in item:
                    url = item["image_url"].get("url", "")
                    if url.startswith("data:") and not url.startswith("data:image/"):
                        logger.debug("Anthropic: converting non-image image_url to file content format")
                        new_content.append(
                            {
                                "type": VisionMessageType.FILE,
                                "file": {"file_data": url},
                            }
                        )
                    else:
                        new_content.append(item)
                else:
                    new_content.append(item)

            message["content"] = new_content

        return messages

    def get_messages(self, prompt, input_data) -> list[dict]:
        """
        Format messages and convert non-image files to Anthropic file content format.
        """
        messages = super().get_messages(prompt, input_data)
        return self._convert_non_image_to_file_content(messages)

    def update_completion_params(self, params: dict[str, Any]) -> dict[str, Any]:
        """Attach Anthropic prompt caching configuration to completion params."""
        params = super().update_completion_params(params)
        if self.cache_control:
            params.setdefault("cache_control_injection_points", []).append(
                {
                    "location": "message",
                    "index": self.cache_control.cache_injection_point_index,
                    "control": self.cache_control.model_dump(
                        exclude_none=True,
                        exclude={"cache_injection_point_index"},
                    ),
                }
            )
        return params

    def _to_strict_function(self, fn: dict) -> dict:
        """Clean one tool's schema to Anthropic's strict subset and attach ``strict``.

        Delegates to :func:`to_strict_subset_function` (optionality via ``required``
        omission, free-form objects → JSON strings, ``additionalProperties: false``)
        and attaches ``strict: true``, which the LiteLLM patch above forwards to
        Anthropic. See :meth:`BaseLLM.transform_tool_schemas` for the shared gating,
        whitelist, and per-request cap (:attr:`MAX_STRICT_TOOLS`).
        """
        return to_strict_subset_function(fn)

__init__(**kwargs)

Initialize the Anthropic LLM node.

Parameters:

Name	Type	Description	Default
**kwargs		Additional keyword arguments.	{}

Source code in dynamiq/nodes/llms/anthropic.py

def __init__(self, **kwargs):
    """Initialize the Anthropic LLM node.

    Args:
        **kwargs: Additional keyword arguments.
    """
    if kwargs.get("client") is None and kwargs.get("connection") is None:
        kwargs["connection"] = AnthropicConnection()
    super().__init__(**kwargs)

get_messages(prompt, input_data)

Format messages and convert non-image files to Anthropic file content format.

Source code in dynamiq/nodes/llms/anthropic.py

def get_messages(self, prompt, input_data) -> list[dict]:
    """
    Format messages and convert non-image files to Anthropic file content format.
    """
    messages = super().get_messages(prompt, input_data)
    return self._convert_non_image_to_file_content(messages)

update_completion_params(params)

Attach Anthropic prompt caching configuration to completion params.

Source code in dynamiq/nodes/llms/anthropic.py

def update_completion_params(self, params: dict[str, Any]) -> dict[str, Any]:
    """Attach Anthropic prompt caching configuration to completion params."""
    params = super().update_completion_params(params)
    if self.cache_control:
        params.setdefault("cache_control_injection_points", []).append(
            {
                "location": "message",
                "index": self.cache_control.cache_injection_point_index,
                "control": self.cache_control.model_dump(
                    exclude_none=True,
                    exclude={"cache_injection_point_index"},
                ),
            }
        )
    return params

AnthropicCacheControl

Bases: BaseModel

Anthropic prompt caching configuration.

Source code in dynamiq/nodes/llms/anthropic.py

class AnthropicCacheControl(BaseModel):
    """Anthropic prompt caching configuration."""

    type: Literal["ephemeral"] = "ephemeral"
    ttl: Literal["5m", "1h"] | None = "5m"
    cache_injection_point_index: int = -2