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 | Literal[False] | None	Prompt caching config. None (the default) chooses nothing, leaving an :class:Agent free to enable caching on this node at construction; False opts out. A bare node caches only when given a config -- None and False send no breakpoints.
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 | Literal[False] | None): Prompt caching config.
            ``None`` (the default) chooses nothing, leaving an :class:`Agent` free to
            enable caching on this node at construction; ``False`` opts out. A bare node
            caches only when given a config -- ``None`` and ``False`` send no breakpoints.
        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
    # ``None`` = nothing chosen (survives a YAML round trip, lets an Agent inject the
    # default); ``False`` = opt out.
    cache_control: AnthropicCacheControl | Literal[False] | 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 the node's own prompt caching configuration to completion params."""
        params = super().update_completion_params(params)
        return self._apply_cache_control(params, self.cache_control)

    def _apply_cache_control(self, params: dict[str, Any], cache_control: Any) -> dict[str, Any]:
        """Attach Anthropic prompt caching breakpoints for the given configuration."""
        if not cache_control:
            return params

        control = cache_control.model_dump(
            exclude_none=True,
            exclude={"cache_injection_point_index", "cache_system"},
        )
        points = params.setdefault("cache_control_injection_points", [])

        # Head first: points are honored in order, so the durable one wins if the
        # 4-block budget runs short. Separate copies -- LiteLLM assigns by reference.
        if cache_control.cache_system:
            points.append({"location": "message", "role": "system", "control": dict(control)})

        points.append(
            {
                "location": "message",
                "index": cache_control.cache_injection_point_index,
                "control": dict(control),
            }
        )
        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 the node's own 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 the node's own prompt caching configuration to completion params."""
    params = super().update_completion_params(params)
    return self._apply_cache_control(params, self.cache_control)

AnthropicCacheControl

Bases: BaseModel

Anthropic prompt caching configuration.

A breakpoint caches everything before it. Anthropic renders the request as tools -> system -> messages, so one point on the system message also covers the tool schemas.

Attributes:

Name	Type	Description
ttl	Literal['5m', '1h'] | None	Cache lifetime for both breakpoints.
cache_injection_point_index	int	Message index for the rolling breakpoint. -1 marks the last message, which is what the next agent loop reads back, so each call writes only the delta.
cache_system	bool	Also pin a breakpoint on the system message, so the system prompt and tool schemas stay cached when the message tail is rewritten (history compaction). Resolves to nothing without a system message.

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

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

    A breakpoint caches everything before it. Anthropic renders the request as
    ``tools -> system -> messages``, so one point on the system message also
    covers the tool schemas.

    Attributes:
        ttl: Cache lifetime for both breakpoints.
        cache_injection_point_index: Message index for the rolling breakpoint.
            ``-1`` marks the last message, which is what the next agent loop
            reads back, so each call writes only the delta.
        cache_system: Also pin a breakpoint on the system message, so the system
            prompt and tool schemas stay cached when the message tail is rewritten
            (history compaction). Resolves to nothing without a system message.
    """

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