`strands.models.llamacpp` ¶

llama.cpp model provider.

Provides integration with llama.cpp servers running in OpenAI-compatible mode, with support for advanced llama.cpp-specific features.

Docs: https://github.com/ggml-org/llama.cpp
Server docs: https://github.com/ggml-org/llama.cpp/tree/master/tools/server
OpenAI API compatibility: https://github.com/ggml-org/llama.cpp/blob/master/tools/server/README.md#api-endpoints

`Messages = List[Message]` `module-attribute` ¶

A list of messages representing a conversation.

`T = TypeVar('T', bound=BaseModel)` `module-attribute` ¶

`ToolChoice = Union[ToolChoiceAutoDict, ToolChoiceAnyDict, ToolChoiceToolDict]` `module-attribute` ¶

Configuration for how the model should choose tools.

"auto": The model decides whether to use tools based on the context
"any": The model must use at least one tool (any tool)
"tool": The model must use the specified tool

`logger = logging.getLogger(name)` `module-attribute` ¶

`ContentBlock` ¶

Bases: TypedDict

A block of content for a message that you pass to, or receive from, a model.

Attributes:

Name	Type	Description
`cachePoint`	`CachePoint`	A cache point configuration to optimize conversation history.
`document`	`DocumentContent`	A document to include in the message.
`guardContent`	`GuardContent`	Contains the content to assess with the guardrail.
`image`	`ImageContent`	Image to include in the message.
`reasoningContent`	`ReasoningContentBlock`	Contains content regarding the reasoning that is carried out by the model.
`text`	`str`	Text to include in the message.
`toolResult`	`ToolResult`	The result for a tool request that a model makes.
`toolUse`	`ToolUse`	Information about a tool use request from a model.
`video`	`VideoContent`	Video to include in the message.
`citationsContent`	`CitationsContentBlock`	Contains the citations for a document.

Source code in strands/types/content.py

class ContentBlock(TypedDict, total=False):
    """A block of content for a message that you pass to, or receive from, a model.

    Attributes:
        cachePoint: A cache point configuration to optimize conversation history.
        document: A document to include in the message.
        guardContent: Contains the content to assess with the guardrail.
        image: Image to include in the message.
        reasoningContent: Contains content regarding the reasoning that is carried out by the model.
        text: Text to include in the message.
        toolResult: The result for a tool request that a model makes.
        toolUse: Information about a tool use request from a model.
        video: Video to include in the message.
        citationsContent: Contains the citations for a document.
    """

    cachePoint: CachePoint
    document: DocumentContent
    guardContent: GuardContent
    image: ImageContent
    reasoningContent: ReasoningContentBlock
    text: str
    toolResult: ToolResult
    toolUse: ToolUse
    video: VideoContent
    citationsContent: CitationsContentBlock

`ContextWindowOverflowException` ¶

Bases: Exception

Exception raised when the context window is exceeded.

This exception is raised when the input to a model exceeds the maximum context window size that the model can handle. This typically occurs when the combined length of the conversation history, system prompt, and current message is too large for the model to process.

Source code in strands/types/exceptions.py

class ContextWindowOverflowException(Exception):
    """Exception raised when the context window is exceeded.

    This exception is raised when the input to a model exceeds the maximum context window size that the model can
    handle. This typically occurs when the combined length of the conversation history, system prompt, and current
    message is too large for the model to process.
    """

    pass

`LlamaCppModel` ¶

Bases: Model

llama.cpp model provider implementation.

Connects to a llama.cpp server running in OpenAI-compatible mode with support for advanced llama.cpp-specific features like grammar constraints, Mirostat sampling, native JSON schema validation, and native multimodal support for audio and image content.

The llama.cpp server must be started with the OpenAI-compatible API enabled: llama-server -m model.gguf --host 0.0.0.0 --port 8080

Example

Basic usage:

model = LlamaCppModel(base_url="http://localhost:8080") model.update_config(params={"temperature": 0.7, "top_k": 40})

Grammar constraints via params:

model.update_config(params={ ... "grammar": ''' ... root ::= answer ... answer ::= "yes" | "no" ... ''' ... })

Advanced sampling:

model.update_config(params={ ... "mirostat": 2, ... "mirostat_lr": 0.1, ... "tfs_z": 0.95, ... "repeat_penalty": 1.1 ... })

Multimodal usage (requires multimodal model like Qwen2.5-Omni):

Audio analysis¶

audio_content = [{ ... "audio": {"source": {"bytes": audio_bytes}, "format": "wav"}, ... "text": "What do you hear in this audio?" ... }] response = agent(audio_content)

Image analysis¶

image_content = [{ ... "image": {"source": {"bytes": image_bytes}, "format": "png"}, ... "text": "Describe this image" ... }] response = agent(image_content)

Source code in strands/models/llamacpp.py

class LlamaCppModel(Model):
    """llama.cpp model provider implementation.

    Connects to a llama.cpp server running in OpenAI-compatible mode with
    support for advanced llama.cpp-specific features like grammar constraints,
    Mirostat sampling, native JSON schema validation, and native multimodal
    support for audio and image content.

    The llama.cpp server must be started with the OpenAI-compatible API enabled:
        llama-server -m model.gguf --host 0.0.0.0 --port 8080

    Example:
        Basic usage:
        >>> model = LlamaCppModel(base_url="http://localhost:8080")
        >>> model.update_config(params={"temperature": 0.7, "top_k": 40})

        Grammar constraints via params:
        >>> model.update_config(params={
        ...     "grammar": '''
        ...         root ::= answer
        ...         answer ::= "yes" | "no"
        ...     '''
        ... })

        Advanced sampling:
        >>> model.update_config(params={
        ...     "mirostat": 2,
        ...     "mirostat_lr": 0.1,
        ...     "tfs_z": 0.95,
        ...     "repeat_penalty": 1.1
        ... })

        Multimodal usage (requires multimodal model like Qwen2.5-Omni):
        >>> # Audio analysis
        >>> audio_content = [{
        ...     "audio": {"source": {"bytes": audio_bytes}, "format": "wav"},
        ...     "text": "What do you hear in this audio?"
        ... }]
        >>> response = agent(audio_content)

        >>> # Image analysis
        >>> image_content = [{
        ...     "image": {"source": {"bytes": image_bytes}, "format": "png"},
        ...     "text": "Describe this image"
        ... }]
        >>> response = agent(image_content)
    """

    class LlamaCppConfig(TypedDict, total=False):
        """Configuration options for llama.cpp models.

        Attributes:
            model_id: Model identifier for the loaded model in llama.cpp server.
                Default is "default" as llama.cpp typically loads a single model.
            params: Model parameters supporting both OpenAI and llama.cpp-specific options.

                OpenAI-compatible parameters:
                - max_tokens: Maximum number of tokens to generate
                - temperature: Sampling temperature (0.0 to 2.0)
                - top_p: Nucleus sampling parameter (0.0 to 1.0)
                - frequency_penalty: Frequency penalty (-2.0 to 2.0)
                - presence_penalty: Presence penalty (-2.0 to 2.0)
                - stop: List of stop sequences
                - seed: Random seed for reproducibility
                - n: Number of completions to generate
                - logprobs: Include log probabilities in output
                - top_logprobs: Number of top log probabilities to include

                llama.cpp-specific parameters:
                - repeat_penalty: Penalize repeat tokens (1.0 = no penalty)
                - top_k: Top-k sampling (0 = disabled)
                - min_p: Min-p sampling threshold (0.0 to 1.0)
                - typical_p: Typical-p sampling (0.0 to 1.0)
                - tfs_z: Tail-free sampling parameter (0.0 to 1.0)
                - top_a: Top-a sampling parameter
                - mirostat: Mirostat sampling mode (0, 1, or 2)
                - mirostat_lr: Mirostat learning rate
                - mirostat_ent: Mirostat target entropy
                - grammar: GBNF grammar string for constrained generation
                - json_schema: JSON schema for structured output
                - penalty_last_n: Number of tokens to consider for penalties
                - n_probs: Number of probabilities to return per token
                - min_keep: Minimum tokens to keep in sampling
                - ignore_eos: Ignore end-of-sequence token
                - logit_bias: Token ID to bias mapping
                - cache_prompt: Cache the prompt for faster generation
                - slot_id: Slot ID for parallel inference
                - samplers: Custom sampler order
        """

        model_id: str
        params: Optional[dict[str, Any]]

    def __init__(
        self,
        base_url: str = "http://localhost:8080",
        timeout: Optional[Union[float, tuple[float, float]]] = None,
        **model_config: Unpack[LlamaCppConfig],
    ) -> None:
        """Initialize llama.cpp provider instance.

        Args:
            base_url: Base URL for the llama.cpp server.
                Default is "http://localhost:8080" for local server.
            timeout: Request timeout in seconds. Can be float or tuple of
                (connect, read) timeouts.
            **model_config: Configuration options for the llama.cpp model.
        """
        validate_config_keys(model_config, self.LlamaCppConfig)

        # Set default model_id if not provided
        if "model_id" not in model_config:
            model_config["model_id"] = "default"

        self.base_url = base_url.rstrip("/")
        self.config = dict(model_config)
        logger.debug("config=<%s> | initializing", self.config)

        # Configure HTTP client
        if isinstance(timeout, tuple):
            # Convert tuple to httpx.Timeout object
            timeout_obj = httpx.Timeout(
                connect=timeout[0] if len(timeout) > 0 else None,
                read=timeout[1] if len(timeout) > 1 else None,
                write=timeout[2] if len(timeout) > 2 else None,
                pool=timeout[3] if len(timeout) > 3 else None,
            )
        else:
            timeout_obj = httpx.Timeout(timeout or 30.0)

        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            timeout=timeout_obj,
        )

    @override
    def update_config(self, **model_config: Unpack[LlamaCppConfig]) -> None:  # type: ignore[override]
        """Update the llama.cpp model configuration with provided arguments.

        Args:
            **model_config: Configuration overrides.
        """
        validate_config_keys(model_config, self.LlamaCppConfig)
        self.config.update(model_config)

    @override
    def get_config(self) -> LlamaCppConfig:
        """Get the llama.cpp model configuration.

        Returns:
            The llama.cpp model configuration.
        """
        return self.config  # type: ignore[return-value]

    def _format_message_content(self, content: Union[ContentBlock, Dict[str, Any]]) -> dict[str, Any]:
        """Format a content block for llama.cpp.

        Args:
            content: Message content.

        Returns:
            llama.cpp compatible content block.

        Raises:
            TypeError: If the content block type cannot be converted to a compatible format.
        """
        if "document" in content:
            mime_type = mimetypes.types_map.get(f".{content['document']['format']}", "application/octet-stream")
            file_data = base64.b64encode(content["document"]["source"]["bytes"]).decode("utf-8")
            return {
                "file": {
                    "file_data": f"data:{mime_type};base64,{file_data}",
                    "filename": content["document"]["name"],
                },
                "type": "file",
            }

        if "image" in content:
            mime_type = mimetypes.types_map.get(f".{content['image']['format']}", "application/octet-stream")
            image_data = base64.b64encode(content["image"]["source"]["bytes"]).decode("utf-8")
            return {
                "image_url": {
                    "detail": "auto",
                    "format": mime_type,
                    "url": f"data:{mime_type};base64,{image_data}",
                },
                "type": "image_url",
            }

        # Handle audio content (not in standard ContentBlock but supported by llama.cpp)
        if "audio" in content:
            audio_content = cast(Dict[str, Any], content)
            audio_data = base64.b64encode(audio_content["audio"]["source"]["bytes"]).decode("utf-8")
            audio_format = audio_content["audio"].get("format", "wav")
            return {
                "type": "input_audio",
                "input_audio": {"data": audio_data, "format": audio_format},
            }

        if "text" in content:
            return {"text": content["text"], "type": "text"}

        raise TypeError(f"content_type=<{next(iter(content))}> | unsupported type")

    def _format_tool_call(self, tool_use: dict[str, Any]) -> dict[str, Any]:
        """Format a tool call for llama.cpp.

        Args:
            tool_use: Tool use requested by the model.

        Returns:
            llama.cpp compatible tool call.
        """
        return {
            "function": {
                "arguments": json.dumps(tool_use["input"]),
                "name": tool_use["name"],
            },
            "id": tool_use["toolUseId"],
            "type": "function",
        }

    def _format_tool_message(self, tool_result: dict[str, Any]) -> dict[str, Any]:
        """Format a tool message for llama.cpp.

        Args:
            tool_result: Tool result collected from a tool execution.

        Returns:
            llama.cpp compatible tool message.
        """
        contents = [
            {"text": json.dumps(content["json"])} if "json" in content else content
            for content in tool_result["content"]
        ]

        return {
            "role": "tool",
            "tool_call_id": tool_result["toolUseId"],
            "content": [self._format_message_content(content) for content in contents],
        }

    def _format_messages(self, messages: Messages, system_prompt: Optional[str] = None) -> list[dict[str, Any]]:
        """Format messages for llama.cpp.

        Args:
            messages: List of message objects to be processed.
            system_prompt: System prompt to provide context to the model.

        Returns:
            Formatted messages array compatible with llama.cpp.
        """
        formatted_messages: list[dict[str, Any]] = []

        # Add system prompt if provided
        if system_prompt:
            formatted_messages.append({"role": "system", "content": system_prompt})

        for message in messages:
            contents = message["content"]

            formatted_contents = [
                self._format_message_content(content)
                for content in contents
                if not any(block_type in content for block_type in ["toolResult", "toolUse"])
            ]
            formatted_tool_calls = [
                self._format_tool_call(
                    {
                        "name": content["toolUse"]["name"],
                        "input": content["toolUse"]["input"],
                        "toolUseId": content["toolUse"]["toolUseId"],
                    }
                )
                for content in contents
                if "toolUse" in content
            ]
            formatted_tool_messages = [
                self._format_tool_message(
                    {
                        "toolUseId": content["toolResult"]["toolUseId"],
                        "content": content["toolResult"]["content"],
                    }
                )
                for content in contents
                if "toolResult" in content
            ]

            formatted_message = {
                "role": message["role"],
                "content": formatted_contents,
                **({} if not formatted_tool_calls else {"tool_calls": formatted_tool_calls}),
            }
            formatted_messages.append(formatted_message)
            formatted_messages.extend(formatted_tool_messages)

        return [message for message in formatted_messages if message["content"] or "tool_calls" in message]

    def _format_request(
        self,
        messages: Messages,
        tool_specs: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
    ) -> dict[str, Any]:
        """Format a request for the llama.cpp server.

        Args:
            messages: List of message objects to be processed by the model.
            tool_specs: List of tool specifications to make available to the model.
            system_prompt: System prompt to provide context to the model.

        Returns:
            A request formatted for llama.cpp server's OpenAI-compatible API.
        """
        # Separate OpenAI-compatible and llama.cpp-specific parameters
        request = {
            "messages": self._format_messages(messages, system_prompt),
            "model": self.config["model_id"],
            "stream": True,
            "stream_options": {"include_usage": True},
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": tool_spec["name"],
                        "description": tool_spec["description"],
                        "parameters": tool_spec["inputSchema"]["json"],
                    },
                }
                for tool_spec in tool_specs or []
            ],
        }

        # Handle parameters if provided
        params = self.config.get("params")
        if params and isinstance(params, dict):
            # Grammar and json_schema go directly in request body for llama.cpp server
            if "grammar" in params:
                request["grammar"] = params["grammar"]
            if "json_schema" in params:
                request["json_schema"] = params["json_schema"]

            # llama.cpp-specific parameters that must be passed via extra_body
            # NOTE: grammar and json_schema are NOT in this set because llama.cpp server
            # expects them directly in the request body for proper constraint application
            llamacpp_specific_params = {
                "repeat_penalty",
                "top_k",
                "min_p",
                "typical_p",
                "tfs_z",
                "top_a",
                "mirostat",
                "mirostat_lr",
                "mirostat_ent",
                "penalty_last_n",
                "n_probs",
                "min_keep",
                "ignore_eos",
                "logit_bias",
                "cache_prompt",
                "slot_id",
                "samplers",
            }

            # Standard OpenAI parameters that go directly in the request
            openai_params = {
                "temperature",
                "max_tokens",
                "top_p",
                "frequency_penalty",
                "presence_penalty",
                "stop",
                "seed",
                "n",
                "logprobs",
                "top_logprobs",
                "response_format",
            }

            # Add OpenAI parameters directly to request
            for param, value in params.items():
                if param in openai_params:
                    request[param] = value

            # Collect llama.cpp-specific parameters for extra_body
            extra_body: Dict[str, Any] = {}
            for param, value in params.items():
                if param in llamacpp_specific_params:
                    extra_body[param] = value

            # Add extra_body if we have llama.cpp-specific parameters
            if extra_body:
                request["extra_body"] = extra_body

        return request

    def _format_chunk(self, event: dict[str, Any]) -> StreamEvent:
        """Format a llama.cpp response event into a standardized message chunk.

        Args:
            event: A response event from the llama.cpp server.

        Returns:
            The formatted chunk.

        Raises:
            RuntimeError: If chunk_type is not recognized.
        """
        match event["chunk_type"]:
            case "message_start":
                return {"messageStart": {"role": "assistant"}}

            case "content_start":
                if event["data_type"] == "tool":
                    return {
                        "contentBlockStart": {
                            "start": {
                                "toolUse": {
                                    "name": event["data"].function.name,
                                    "toolUseId": event["data"].id,
                                }
                            }
                        }
                    }
                return {"contentBlockStart": {"start": {}}}

            case "content_delta":
                if event["data_type"] == "tool":
                    return {
                        "contentBlockDelta": {"delta": {"toolUse": {"input": event["data"].function.arguments or ""}}}
                    }
                if event["data_type"] == "reasoning_content":
                    return {"contentBlockDelta": {"delta": {"reasoningContent": {"text": event["data"]}}}}
                return {"contentBlockDelta": {"delta": {"text": event["data"]}}}

            case "content_stop":
                return {"contentBlockStop": {}}

            case "message_stop":
                match event["data"]:
                    case "tool_calls":
                        return {"messageStop": {"stopReason": "tool_use"}}
                    case "length":
                        return {"messageStop": {"stopReason": "max_tokens"}}
                    case _:
                        return {"messageStop": {"stopReason": "end_turn"}}

            case "metadata":
                return {
                    "metadata": {
                        "usage": {
                            "inputTokens": event["data"].prompt_tokens,
                            "outputTokens": event["data"].completion_tokens,
                            "totalTokens": event["data"].total_tokens,
                        },
                        "metrics": {
                            "latencyMs": event.get("latency_ms", 0),
                        },
                    },
                }

            case _:
                raise RuntimeError(f"chunk_type=<{event['chunk_type']}> | unknown type")

    @override
    async def stream(
        self,
        messages: Messages,
        tool_specs: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
        *,
        tool_choice: ToolChoice | None = None,
        **kwargs: Any,
    ) -> AsyncGenerator[StreamEvent, None]:
        """Stream conversation with the llama.cpp model.

        Args:
            messages: List of message objects to be processed by the model.
            tool_specs: List of tool specifications to make available to the model.
            system_prompt: System prompt to provide context to the model.
            tool_choice: Selection strategy for tool invocation. **Note: This parameter is accepted for
                interface consistency but is currently ignored for this model provider.**
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Formatted message chunks from the model.

        Raises:
            ContextWindowOverflowException: When the context window is exceeded.
            ModelThrottledException: When the llama.cpp server is overloaded.
        """
        warn_on_tool_choice_not_supported(tool_choice)

        # Track request start time for latency calculation
        start_time = time.perf_counter()

        try:
            logger.debug("formatting request")
            request = self._format_request(messages, tool_specs, system_prompt)
            logger.debug("request=<%s>", request)

            logger.debug("invoking model")
            response = await self.client.post("/v1/chat/completions", json=request)
            response.raise_for_status()

            logger.debug("got response from model")
            yield self._format_chunk({"chunk_type": "message_start"})
            yield self._format_chunk({"chunk_type": "content_start", "data_type": "text"})

            tool_calls: Dict[int, list] = {}
            usage_data = None
            finish_reason = None

            async for line in response.aiter_lines():
                if not line.strip() or not line.startswith("data: "):
                    continue

                data_content = line[6:]  # Remove "data: " prefix
                if data_content.strip() == "[DONE]":
                    break

                try:
                    event = json.loads(data_content)
                except json.JSONDecodeError:
                    continue

                # Handle usage information
                if "usage" in event:
                    usage_data = event["usage"]
                    continue

                if not event.get("choices"):
                    continue

                choice = event["choices"][0]
                delta = choice.get("delta", {})

                # Handle content deltas
                if "content" in delta and delta["content"]:
                    yield self._format_chunk(
                        {
                            "chunk_type": "content_delta",
                            "data_type": "text",
                            "data": delta["content"],
                        }
                    )

                # Handle tool calls
                if "tool_calls" in delta:
                    for tool_call in delta["tool_calls"]:
                        index = tool_call["index"]
                        if index not in tool_calls:
                            tool_calls[index] = []
                        tool_calls[index].append(tool_call)

                # Check for finish reason
                if choice.get("finish_reason"):
                    finish_reason = choice.get("finish_reason")
                    break

            yield self._format_chunk({"chunk_type": "content_stop"})

            # Process tool calls
            for tool_deltas in tool_calls.values():
                first_delta = tool_deltas[0]
                yield self._format_chunk(
                    {
                        "chunk_type": "content_start",
                        "data_type": "tool",
                        "data": type(
                            "ToolCall",
                            (),
                            {
                                "function": type(
                                    "Function",
                                    (),
                                    {
                                        "name": first_delta.get("function", {}).get("name", ""),
                                    },
                                )(),
                                "id": first_delta.get("id", ""),
                            },
                        )(),
                    }
                )

                for tool_delta in tool_deltas:
                    yield self._format_chunk(
                        {
                            "chunk_type": "content_delta",
                            "data_type": "tool",
                            "data": type(
                                "ToolCall",
                                (),
                                {
                                    "function": type(
                                        "Function",
                                        (),
                                        {
                                            "arguments": tool_delta.get("function", {}).get("arguments", ""),
                                        },
                                    )(),
                                },
                            )(),
                        }
                    )

                yield self._format_chunk({"chunk_type": "content_stop"})

            # Send stop reason
            if finish_reason == "tool_calls" or tool_calls:
                stop_reason = "tool_calls"  # Changed from "tool_use" to match format_chunk expectations
            else:
                stop_reason = finish_reason or "end_turn"
            yield self._format_chunk({"chunk_type": "message_stop", "data": stop_reason})

            # Send usage metadata if available
            if usage_data:
                # Calculate latency
                latency_ms = int((time.perf_counter() - start_time) * 1000)
                yield self._format_chunk(
                    {
                        "chunk_type": "metadata",
                        "data": type(
                            "Usage",
                            (),
                            {
                                "prompt_tokens": usage_data.get("prompt_tokens", 0),
                                "completion_tokens": usage_data.get("completion_tokens", 0),
                                "total_tokens": usage_data.get("total_tokens", 0),
                            },
                        )(),
                        "latency_ms": latency_ms,
                    }
                )

            logger.debug("finished streaming response from model")

        except httpx.HTTPStatusError as e:
            if e.response.status_code == 400:
                # Parse error response from llama.cpp server
                try:
                    error_data = e.response.json()
                    error_msg = str(error_data.get("error", {}).get("message", str(error_data)))
                except (json.JSONDecodeError, KeyError, AttributeError):
                    error_msg = e.response.text

                # Check for context overflow by looking for specific error indicators
                if any(term in error_msg.lower() for term in ["context", "kv cache", "slot"]):
                    raise ContextWindowOverflowException(f"Context window exceeded: {error_msg}") from e
            elif e.response.status_code == 503:
                raise ModelThrottledException("llama.cpp server is busy or overloaded") from e
            raise
        except Exception as e:
            # Handle other potential errors like rate limiting
            error_msg = str(e).lower()
            if "rate" in error_msg or "429" in str(e):
                raise ModelThrottledException(str(e)) from e
            raise

    @override
    async def structured_output(
        self,
        output_model: Type[T],
        prompt: Messages,
        system_prompt: Optional[str] = None,
        **kwargs: Any,
    ) -> AsyncGenerator[dict[str, Union[T, Any]], None]:
        """Get structured output using llama.cpp's native JSON schema support.

        This implementation uses llama.cpp's json_schema parameter to constrain
        the model output to valid JSON matching the provided schema.

        Args:
            output_model: The Pydantic model defining the expected output structure.
            prompt: The prompt messages to use for generation.
            system_prompt: System prompt to provide context to the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Model events with the last being the structured output.

        Raises:
            json.JSONDecodeError: If the model output is not valid JSON.
            pydantic.ValidationError: If the output doesn't match the model schema.
        """
        # Get the JSON schema from the Pydantic model
        schema = output_model.model_json_schema()

        # Store current params to restore later
        params = self.config.get("params", {})
        original_params = dict(params) if isinstance(params, dict) else {}

        try:
            # Configure for JSON output with schema constraint
            params = self.config.get("params", {})
            if not isinstance(params, dict):
                params = {}
            params["json_schema"] = schema
            params["cache_prompt"] = True
            self.config["params"] = params

            # Collect the response
            response_text = ""
            async for event in self.stream(prompt, system_prompt=system_prompt, **kwargs):
                if "contentBlockDelta" in event:
                    delta = event["contentBlockDelta"]["delta"]
                    if "text" in delta:
                        response_text += delta["text"]
                # Forward events to caller
                yield cast(Dict[str, Union[T, Any]], event)

            # Parse and validate the JSON response
            data = json.loads(response_text.strip())
            output_instance = output_model(**data)
            yield {"output": output_instance}

        finally:
            # Restore original configuration
            self.config["params"] = original_params

`LlamaCppConfig` ¶

Bases: TypedDict

Configuration options for llama.cpp models.

Attributes:

Name	Type	Description
`model_id`	`str`	Model identifier for the loaded model in llama.cpp server. Default is "default" as llama.cpp typically loads a single model.
`params`	`Optional[dict[str, Any]]`	Model parameters supporting both OpenAI and llama.cpp-specific options. OpenAI-compatible parameters: - max_tokens: Maximum number of tokens to generate - temperature: Sampling temperature (0.0 to 2.0) - top_p: Nucleus sampling parameter (0.0 to 1.0) - frequency_penalty: Frequency penalty (-2.0 to 2.0) - presence_penalty: Presence penalty (-2.0 to 2.0) - stop: List of stop sequences - seed: Random seed for reproducibility - n: Number of completions to generate - logprobs: Include log probabilities in output - top_logprobs: Number of top log probabilities to include llama.cpp-specific parameters: - repeat_penalty: Penalize repeat tokens (1.0 = no penalty) - top_k: Top-k sampling (0 = disabled) - min_p: Min-p sampling threshold (0.0 to 1.0) - typical_p: Typical-p sampling (0.0 to 1.0) - tfs_z: Tail-free sampling parameter (0.0 to 1.0) - top_a: Top-a sampling parameter - mirostat: Mirostat sampling mode (0, 1, or 2) - mirostat_lr: Mirostat learning rate - mirostat_ent: Mirostat target entropy - grammar: GBNF grammar string for constrained generation - json_schema: JSON schema for structured output - penalty_last_n: Number of tokens to consider for penalties - n_probs: Number of probabilities to return per token - min_keep: Minimum tokens to keep in sampling - ignore_eos: Ignore end-of-sequence token - logit_bias: Token ID to bias mapping - cache_prompt: Cache the prompt for faster generation - slot_id: Slot ID for parallel inference - samplers: Custom sampler order

Source code in strands/models/llamacpp.py

class LlamaCppConfig(TypedDict, total=False):
    """Configuration options for llama.cpp models.

    Attributes:
        model_id: Model identifier for the loaded model in llama.cpp server.
            Default is "default" as llama.cpp typically loads a single model.
        params: Model parameters supporting both OpenAI and llama.cpp-specific options.

            OpenAI-compatible parameters:
            - max_tokens: Maximum number of tokens to generate
            - temperature: Sampling temperature (0.0 to 2.0)
            - top_p: Nucleus sampling parameter (0.0 to 1.0)
            - frequency_penalty: Frequency penalty (-2.0 to 2.0)
            - presence_penalty: Presence penalty (-2.0 to 2.0)
            - stop: List of stop sequences
            - seed: Random seed for reproducibility
            - n: Number of completions to generate
            - logprobs: Include log probabilities in output
            - top_logprobs: Number of top log probabilities to include

            llama.cpp-specific parameters:
            - repeat_penalty: Penalize repeat tokens (1.0 = no penalty)
            - top_k: Top-k sampling (0 = disabled)
            - min_p: Min-p sampling threshold (0.0 to 1.0)
            - typical_p: Typical-p sampling (0.0 to 1.0)
            - tfs_z: Tail-free sampling parameter (0.0 to 1.0)
            - top_a: Top-a sampling parameter
            - mirostat: Mirostat sampling mode (0, 1, or 2)
            - mirostat_lr: Mirostat learning rate
            - mirostat_ent: Mirostat target entropy
            - grammar: GBNF grammar string for constrained generation
            - json_schema: JSON schema for structured output
            - penalty_last_n: Number of tokens to consider for penalties
            - n_probs: Number of probabilities to return per token
            - min_keep: Minimum tokens to keep in sampling
            - ignore_eos: Ignore end-of-sequence token
            - logit_bias: Token ID to bias mapping
            - cache_prompt: Cache the prompt for faster generation
            - slot_id: Slot ID for parallel inference
            - samplers: Custom sampler order
    """

    model_id: str
    params: Optional[dict[str, Any]]

`init(base_url='http://localhost:8080', timeout=None, **model_config)` ¶

Initialize llama.cpp provider instance.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Base URL for the llama.cpp server. Default is "http://localhost:8080" for local server.	`'http://localhost:8080'`
`timeout`	`Optional[Union[float, tuple[float, float]]]`	Request timeout in seconds. Can be float or tuple of (connect, read) timeouts.	`None`
`**model_config`	`Unpack[LlamaCppConfig]`	Configuration options for the llama.cpp model.	`{}`

Source code in strands/models/llamacpp.py

def __init__(
    self,
    base_url: str = "http://localhost:8080",
    timeout: Optional[Union[float, tuple[float, float]]] = None,
    **model_config: Unpack[LlamaCppConfig],
) -> None:
    """Initialize llama.cpp provider instance.

    Args:
        base_url: Base URL for the llama.cpp server.
            Default is "http://localhost:8080" for local server.
        timeout: Request timeout in seconds. Can be float or tuple of
            (connect, read) timeouts.
        **model_config: Configuration options for the llama.cpp model.
    """
    validate_config_keys(model_config, self.LlamaCppConfig)

    # Set default model_id if not provided
    if "model_id" not in model_config:
        model_config["model_id"] = "default"

    self.base_url = base_url.rstrip("/")
    self.config = dict(model_config)
    logger.debug("config=<%s> | initializing", self.config)

    # Configure HTTP client
    if isinstance(timeout, tuple):
        # Convert tuple to httpx.Timeout object
        timeout_obj = httpx.Timeout(
            connect=timeout[0] if len(timeout) > 0 else None,
            read=timeout[1] if len(timeout) > 1 else None,
            write=timeout[2] if len(timeout) > 2 else None,
            pool=timeout[3] if len(timeout) > 3 else None,
        )
    else:
        timeout_obj = httpx.Timeout(timeout or 30.0)

    self.client = httpx.AsyncClient(
        base_url=self.base_url,
        timeout=timeout_obj,
    )

`get_config()` ¶

Get the llama.cpp model configuration.

Returns:

Type	Description
`LlamaCppConfig`	The llama.cpp model configuration.

Source code in strands/models/llamacpp.py

@override
def get_config(self) -> LlamaCppConfig:
    """Get the llama.cpp model configuration.

    Returns:
        The llama.cpp model configuration.
    """
    return self.config  # type: ignore[return-value]

`stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, **kwargs)` `async` ¶

Stream conversation with the llama.cpp model.

Parameters:

Name	Type	Description	Default
`messages`	`Messages`	List of message objects to be processed by the model.	required
`tool_specs`	`Optional[list[ToolSpec]]`	List of tool specifications to make available to the model.	`None`
`system_prompt`	`Optional[str]`	System prompt to provide context to the model.	`None`
`tool_choice`	`ToolChoice \| None`	Selection strategy for tool invocation. Note: This parameter is accepted for interface consistency but is currently ignored for this model provider.	`None`
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Yields:

Type	Description
`AsyncGenerator[StreamEvent, None]`	Formatted message chunks from the model.

Raises:

Type	Description
`ContextWindowOverflowException`	When the context window is exceeded.
`ModelThrottledException`	When the llama.cpp server is overloaded.

Source code in strands/models/llamacpp.py

@override
async def stream(
    self,
    messages: Messages,
    tool_specs: Optional[list[ToolSpec]] = None,
    system_prompt: Optional[str] = None,
    *,
    tool_choice: ToolChoice | None = None,
    **kwargs: Any,
) -> AsyncGenerator[StreamEvent, None]:
    """Stream conversation with the llama.cpp model.

    Args:
        messages: List of message objects to be processed by the model.
        tool_specs: List of tool specifications to make available to the model.
        system_prompt: System prompt to provide context to the model.
        tool_choice: Selection strategy for tool invocation. **Note: This parameter is accepted for
            interface consistency but is currently ignored for this model provider.**
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Formatted message chunks from the model.

    Raises:
        ContextWindowOverflowException: When the context window is exceeded.
        ModelThrottledException: When the llama.cpp server is overloaded.
    """
    warn_on_tool_choice_not_supported(tool_choice)

    # Track request start time for latency calculation
    start_time = time.perf_counter()

    try:
        logger.debug("formatting request")
        request = self._format_request(messages, tool_specs, system_prompt)
        logger.debug("request=<%s>", request)

        logger.debug("invoking model")
        response = await self.client.post("/v1/chat/completions", json=request)
        response.raise_for_status()

        logger.debug("got response from model")
        yield self._format_chunk({"chunk_type": "message_start"})
        yield self._format_chunk({"chunk_type": "content_start", "data_type": "text"})

        tool_calls: Dict[int, list] = {}
        usage_data = None
        finish_reason = None

        async for line in response.aiter_lines():
            if not line.strip() or not line.startswith("data: "):
                continue

            data_content = line[6:]  # Remove "data: " prefix
            if data_content.strip() == "[DONE]":
                break

            try:
                event = json.loads(data_content)
            except json.JSONDecodeError:
                continue

            # Handle usage information
            if "usage" in event:
                usage_data = event["usage"]
                continue

            if not event.get("choices"):
                continue

            choice = event["choices"][0]
            delta = choice.get("delta", {})

            # Handle content deltas
            if "content" in delta and delta["content"]:
                yield self._format_chunk(
                    {
                        "chunk_type": "content_delta",
                        "data_type": "text",
                        "data": delta["content"],
                    }
                )

            # Handle tool calls
            if "tool_calls" in delta:
                for tool_call in delta["tool_calls"]:
                    index = tool_call["index"]
                    if index not in tool_calls:
                        tool_calls[index] = []
                    tool_calls[index].append(tool_call)

            # Check for finish reason
            if choice.get("finish_reason"):
                finish_reason = choice.get("finish_reason")
                break

        yield self._format_chunk({"chunk_type": "content_stop"})

        # Process tool calls
        for tool_deltas in tool_calls.values():
            first_delta = tool_deltas[0]
            yield self._format_chunk(
                {
                    "chunk_type": "content_start",
                    "data_type": "tool",
                    "data": type(
                        "ToolCall",
                        (),
                        {
                            "function": type(
                                "Function",
                                (),
                                {
                                    "name": first_delta.get("function", {}).get("name", ""),
                                },
                            )(),
                            "id": first_delta.get("id", ""),
                        },
                    )(),
                }
            )

            for tool_delta in tool_deltas:
                yield self._format_chunk(
                    {
                        "chunk_type": "content_delta",
                        "data_type": "tool",
                        "data": type(
                            "ToolCall",
                            (),
                            {
                                "function": type(
                                    "Function",
                                    (),
                                    {
                                        "arguments": tool_delta.get("function", {}).get("arguments", ""),
                                    },
                                )(),
                            },
                        )(),
                    }
                )

            yield self._format_chunk({"chunk_type": "content_stop"})

        # Send stop reason
        if finish_reason == "tool_calls" or tool_calls:
            stop_reason = "tool_calls"  # Changed from "tool_use" to match format_chunk expectations
        else:
            stop_reason = finish_reason or "end_turn"
        yield self._format_chunk({"chunk_type": "message_stop", "data": stop_reason})

        # Send usage metadata if available
        if usage_data:
            # Calculate latency
            latency_ms = int((time.perf_counter() - start_time) * 1000)
            yield self._format_chunk(
                {
                    "chunk_type": "metadata",
                    "data": type(
                        "Usage",
                        (),
                        {
                            "prompt_tokens": usage_data.get("prompt_tokens", 0),
                            "completion_tokens": usage_data.get("completion_tokens", 0),
                            "total_tokens": usage_data.get("total_tokens", 0),
                        },
                    )(),
                    "latency_ms": latency_ms,
                }
            )

        logger.debug("finished streaming response from model")

    except httpx.HTTPStatusError as e:
        if e.response.status_code == 400:
            # Parse error response from llama.cpp server
            try:
                error_data = e.response.json()
                error_msg = str(error_data.get("error", {}).get("message", str(error_data)))
            except (json.JSONDecodeError, KeyError, AttributeError):
                error_msg = e.response.text

            # Check for context overflow by looking for specific error indicators
            if any(term in error_msg.lower() for term in ["context", "kv cache", "slot"]):
                raise ContextWindowOverflowException(f"Context window exceeded: {error_msg}") from e
        elif e.response.status_code == 503:
            raise ModelThrottledException("llama.cpp server is busy or overloaded") from e
        raise
    except Exception as e:
        # Handle other potential errors like rate limiting
        error_msg = str(e).lower()
        if "rate" in error_msg or "429" in str(e):
            raise ModelThrottledException(str(e)) from e
        raise

`structured_output(output_model, prompt, system_prompt=None, **kwargs)` `async` ¶

Get structured output using llama.cpp's native JSON schema support.

This implementation uses llama.cpp's json_schema parameter to constrain the model output to valid JSON matching the provided schema.

Parameters:

Name	Type	Description	Default
`output_model`	`Type[T]`	The Pydantic model defining the expected output structure.	required
`prompt`	`Messages`	The prompt messages to use for generation.	required
`system_prompt`	`Optional[str]`	System prompt to provide context to the model.	`None`
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Yields:

Type	Description
`AsyncGenerator[dict[str, Union[T, Any]], None]`	Model events with the last being the structured output.

Raises:

Type	Description
`JSONDecodeError`	If the model output is not valid JSON.
`ValidationError`	If the output doesn't match the model schema.

Source code in strands/models/llamacpp.py

@override
async def structured_output(
    self,
    output_model: Type[T],
    prompt: Messages,
    system_prompt: Optional[str] = None,
    **kwargs: Any,
) -> AsyncGenerator[dict[str, Union[T, Any]], None]:
    """Get structured output using llama.cpp's native JSON schema support.

    This implementation uses llama.cpp's json_schema parameter to constrain
    the model output to valid JSON matching the provided schema.

    Args:
        output_model: The Pydantic model defining the expected output structure.
        prompt: The prompt messages to use for generation.
        system_prompt: System prompt to provide context to the model.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Model events with the last being the structured output.

    Raises:
        json.JSONDecodeError: If the model output is not valid JSON.
        pydantic.ValidationError: If the output doesn't match the model schema.
    """
    # Get the JSON schema from the Pydantic model
    schema = output_model.model_json_schema()

    # Store current params to restore later
    params = self.config.get("params", {})
    original_params = dict(params) if isinstance(params, dict) else {}

    try:
        # Configure for JSON output with schema constraint
        params = self.config.get("params", {})
        if not isinstance(params, dict):
            params = {}
        params["json_schema"] = schema
        params["cache_prompt"] = True
        self.config["params"] = params

        # Collect the response
        response_text = ""
        async for event in self.stream(prompt, system_prompt=system_prompt, **kwargs):
            if "contentBlockDelta" in event:
                delta = event["contentBlockDelta"]["delta"]
                if "text" in delta:
                    response_text += delta["text"]
            # Forward events to caller
            yield cast(Dict[str, Union[T, Any]], event)

        # Parse and validate the JSON response
        data = json.loads(response_text.strip())
        output_instance = output_model(**data)
        yield {"output": output_instance}

    finally:
        # Restore original configuration
        self.config["params"] = original_params

`update_config(**model_config)` ¶

Update the llama.cpp model configuration with provided arguments.

Parameters:

Name	Type	Description	Default
`**model_config`	`Unpack[LlamaCppConfig]`	Configuration overrides.	`{}`

Source code in strands/models/llamacpp.py

@override
def update_config(self, **model_config: Unpack[LlamaCppConfig]) -> None:  # type: ignore[override]
    """Update the llama.cpp model configuration with provided arguments.

    Args:
        **model_config: Configuration overrides.
    """
    validate_config_keys(model_config, self.LlamaCppConfig)
    self.config.update(model_config)

`Model` ¶

Bases: ABC

Abstract base class for Agent model providers.

This class defines the interface for all model implementations in the Strands Agents SDK. It provides a standardized way to configure and process requests for different AI model providers.

Source code in strands/models/model.py

class Model(abc.ABC):
    """Abstract base class for Agent model providers.

    This class defines the interface for all model implementations in the Strands Agents SDK. It provides a
    standardized way to configure and process requests for different AI model providers.
    """

    @abc.abstractmethod
    # pragma: no cover
    def update_config(self, **model_config: Any) -> None:
        """Update the model configuration with the provided arguments.

        Args:
            **model_config: Configuration overrides.
        """
        pass

    @abc.abstractmethod
    # pragma: no cover
    def get_config(self) -> Any:
        """Return the model configuration.

        Returns:
            The model's configuration.
        """
        pass

    @abc.abstractmethod
    # pragma: no cover
    def structured_output(
        self, output_model: Type[T], prompt: Messages, system_prompt: Optional[str] = None, **kwargs: Any
    ) -> AsyncGenerator[dict[str, Union[T, Any]], None]:
        """Get structured output from the model.

        Args:
            output_model: The output model to use for the agent.
            prompt: The prompt messages to use for the agent.
            system_prompt: System prompt to provide context to the model.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Model events with the last being the structured output.

        Raises:
            ValidationException: The response format from the model does not match the output_model
        """
        pass

    @abc.abstractmethod
    # pragma: no cover
    def stream(
        self,
        messages: Messages,
        tool_specs: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
        *,
        tool_choice: ToolChoice | None = None,
        system_prompt_content: list[SystemContentBlock] | None = None,
        **kwargs: Any,
    ) -> AsyncIterable[StreamEvent]:
        """Stream conversation with the model.

        This method handles the full lifecycle of conversing with the model:

        1. Format the messages, tool specs, and configuration into a streaming request
        2. Send the request to the model
        3. Yield the formatted message chunks

        Args:
            messages: List of message objects to be processed by the model.
            tool_specs: List of tool specifications to make available to the model.
            system_prompt: System prompt to provide context to the model.
            tool_choice: Selection strategy for tool invocation.
            system_prompt_content: System prompt content blocks for advanced features like caching.
            **kwargs: Additional keyword arguments for future extensibility.

        Yields:
            Formatted message chunks from the model.

        Raises:
            ModelThrottledException: When the model service is throttling requests from the client.
        """
        pass

`get_config()` `abstractmethod` ¶

Return the model configuration.

Returns:

Type	Description
`Any`	The model's configuration.

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def get_config(self) -> Any:
    """Return the model configuration.

    Returns:
        The model's configuration.
    """
    pass

`stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, system_prompt_content=None, **kwargs)` `abstractmethod` ¶

Stream conversation with the model.

This method handles the full lifecycle of conversing with the model:

Format the messages, tool specs, and configuration into a streaming request
Send the request to the model
Yield the formatted message chunks

Parameters:

Name	Type	Description	Default
`messages`	`Messages`	List of message objects to be processed by the model.	required
`tool_specs`	`Optional[list[ToolSpec]]`	List of tool specifications to make available to the model.	`None`
`system_prompt`	`Optional[str]`	System prompt to provide context to the model.	`None`
`tool_choice`	`ToolChoice \| None`	Selection strategy for tool invocation.	`None`
`system_prompt_content`	`list[SystemContentBlock] \| None`	System prompt content blocks for advanced features like caching.	`None`
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Yields:

Type	Description
`AsyncIterable[StreamEvent]`	Formatted message chunks from the model.

Raises:

Type	Description
`ModelThrottledException`	When the model service is throttling requests from the client.

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def stream(
    self,
    messages: Messages,
    tool_specs: Optional[list[ToolSpec]] = None,
    system_prompt: Optional[str] = None,
    *,
    tool_choice: ToolChoice | None = None,
    system_prompt_content: list[SystemContentBlock] | None = None,
    **kwargs: Any,
) -> AsyncIterable[StreamEvent]:
    """Stream conversation with the model.

    This method handles the full lifecycle of conversing with the model:

    1. Format the messages, tool specs, and configuration into a streaming request
    2. Send the request to the model
    3. Yield the formatted message chunks

    Args:
        messages: List of message objects to be processed by the model.
        tool_specs: List of tool specifications to make available to the model.
        system_prompt: System prompt to provide context to the model.
        tool_choice: Selection strategy for tool invocation.
        system_prompt_content: System prompt content blocks for advanced features like caching.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Formatted message chunks from the model.

    Raises:
        ModelThrottledException: When the model service is throttling requests from the client.
    """
    pass

`structured_output(output_model, prompt, system_prompt=None, **kwargs)` `abstractmethod` ¶

Get structured output from the model.

Parameters:

Name	Type	Description	Default
`output_model`	`Type[T]`	The output model to use for the agent.	required
`prompt`	`Messages`	The prompt messages to use for the agent.	required
`system_prompt`	`Optional[str]`	System prompt to provide context to the model.	`None`
`**kwargs`	`Any`	Additional keyword arguments for future extensibility.	`{}`

Yields:

Type	Description
`AsyncGenerator[dict[str, Union[T, Any]], None]`	Model events with the last being the structured output.

Raises:

Type	Description
`ValidationException`	The response format from the model does not match the output_model

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def structured_output(
    self, output_model: Type[T], prompt: Messages, system_prompt: Optional[str] = None, **kwargs: Any
) -> AsyncGenerator[dict[str, Union[T, Any]], None]:
    """Get structured output from the model.

    Args:
        output_model: The output model to use for the agent.
        prompt: The prompt messages to use for the agent.
        system_prompt: System prompt to provide context to the model.
        **kwargs: Additional keyword arguments for future extensibility.

    Yields:
        Model events with the last being the structured output.

    Raises:
        ValidationException: The response format from the model does not match the output_model
    """
    pass

`update_config(**model_config)` `abstractmethod` ¶

Update the model configuration with the provided arguments.

Parameters:

Name	Type	Description	Default
`**model_config`	`Any`	Configuration overrides.	`{}`

Source code in strands/models/model.py

@abc.abstractmethod
# pragma: no cover
def update_config(self, **model_config: Any) -> None:
    """Update the model configuration with the provided arguments.

    Args:
        **model_config: Configuration overrides.
    """
    pass

`ModelThrottledException` ¶

Bases: Exception

Exception raised when the model is throttled.

This exception is raised when the model is throttled by the service. This typically occurs when the service is throttling the requests from the client.

Source code in strands/types/exceptions.py

class ModelThrottledException(Exception):
    """Exception raised when the model is throttled.

    This exception is raised when the model is throttled by the service. This typically occurs when the service is
    throttling the requests from the client.
    """

    def __init__(self, message: str) -> None:
        """Initialize exception.

        Args:
            message: The message from the service that describes the throttling.
        """
        self.message = message
        super().__init__(message)

    pass

`init(message)` ¶

Initialize exception.

Parameters:

Name	Type	Description	Default
`message`	`str`	The message from the service that describes the throttling.	required

Source code in strands/types/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize exception.

    Args:
        message: The message from the service that describes the throttling.
    """
    self.message = message
    super().__init__(message)

`StreamEvent` ¶

Bases: TypedDict

The messages output stream.

Attributes:

Name	Type	Description
`contentBlockDelta`	`ContentBlockDeltaEvent`	Delta content for a content block.
`contentBlockStart`	`ContentBlockStartEvent`	Start of a content block.
`contentBlockStop`	`ContentBlockStopEvent`	End of a content block.
`internalServerException`	`ExceptionEvent`	Internal server error information.
`messageStart`	`MessageStartEvent`	Start of a message.
`messageStop`	`MessageStopEvent`	End of a message.
`metadata`	`MetadataEvent`	Metadata about the streaming response.
`modelStreamErrorException`	`ModelStreamErrorEvent`	Model streaming error information.
`serviceUnavailableException`	`ExceptionEvent`	Service unavailable error information.
`throttlingException`	`ExceptionEvent`	Throttling error information.
`validationException`	`ExceptionEvent`	Validation error information.

Source code in strands/types/streaming.py

class StreamEvent(TypedDict, total=False):
    """The messages output stream.

    Attributes:
        contentBlockDelta: Delta content for a content block.
        contentBlockStart: Start of a content block.
        contentBlockStop: End of a content block.
        internalServerException: Internal server error information.
        messageStart: Start of a message.
        messageStop: End of a message.
        metadata: Metadata about the streaming response.
        modelStreamErrorException: Model streaming error information.
        serviceUnavailableException: Service unavailable error information.
        throttlingException: Throttling error information.
        validationException: Validation error information.
    """

    contentBlockDelta: ContentBlockDeltaEvent
    contentBlockStart: ContentBlockStartEvent
    contentBlockStop: ContentBlockStopEvent
    internalServerException: ExceptionEvent
    messageStart: MessageStartEvent
    messageStop: MessageStopEvent
    metadata: MetadataEvent
    redactContent: RedactContentEvent
    modelStreamErrorException: ModelStreamErrorEvent
    serviceUnavailableException: ExceptionEvent
    throttlingException: ExceptionEvent
    validationException: ExceptionEvent

`ToolSpec` ¶

Bases: TypedDict

Specification for a tool that can be used by an agent.

Attributes:

Name	Type	Description
`description`	`str`	A human-readable description of what the tool does.
`inputSchema`	`JSONSchema`	JSON Schema defining the expected input parameters.
`name`	`str`	The unique name of the tool.
`outputSchema`	`NotRequired[JSONSchema]`	Optional JSON Schema defining the expected output format. Note: Not all model providers support this field. Providers that don't support it should filter it out before sending to their API.

Source code in strands/types/tools.py

class ToolSpec(TypedDict):
    """Specification for a tool that can be used by an agent.

    Attributes:
        description: A human-readable description of what the tool does.
        inputSchema: JSON Schema defining the expected input parameters.
        name: The unique name of the tool.
        outputSchema: Optional JSON Schema defining the expected output format.
            Note: Not all model providers support this field. Providers that don't
            support it should filter it out before sending to their API.
    """

    description: str
    inputSchema: JSONSchema
    name: str
    outputSchema: NotRequired[JSONSchema]

`validate_config_keys(config_dict, config_class)` ¶

Validate that config keys match the TypedDict fields.

Parameters:

Name	Type	Description	Default
`config_dict`	`Mapping[str, Any]`	Dictionary of configuration parameters	required
`config_class`	`Type`	TypedDict class to validate against	required

Source code in strands/models/_validation.py

def validate_config_keys(config_dict: Mapping[str, Any], config_class: Type) -> None:
    """Validate that config keys match the TypedDict fields.

    Args:
        config_dict: Dictionary of configuration parameters
        config_class: TypedDict class to validate against
    """
    valid_keys = set(get_type_hints(config_class).keys())
    provided_keys = set(config_dict.keys())
    invalid_keys = provided_keys - valid_keys

    if invalid_keys:
        warnings.warn(
            f"Invalid configuration parameters: {sorted(invalid_keys)}."
            f"\nValid parameters are: {sorted(valid_keys)}."
            f"\n"
            f"\nSee https://github.com/strands-agents/sdk-python/issues/815",
            stacklevel=4,
        )

`warn_on_tool_choice_not_supported(tool_choice)` ¶

Emits a warning if a tool choice is provided but not supported by the provider.

Parameters:

Name	Type	Description	Default
`tool_choice`	`ToolChoice \| None`	the tool_choice provided to the provider	required

Source code in strands/models/_validation.py

def warn_on_tool_choice_not_supported(tool_choice: ToolChoice | None) -> None:
    """Emits a warning if a tool choice is provided but not supported by the provider.

    Args:
        tool_choice: the tool_choice provided to the provider
    """
    if tool_choice:
        warnings.warn(
            "A ToolChoice was provided to this provider but is not supported and will be ignored",
            stacklevel=4,
        )

strands.models.llamacpp ¶

Messages = List[Message] module-attribute ¶

T = TypeVar('T', bound=BaseModel) module-attribute ¶

ToolChoice = Union[ToolChoiceAutoDict, ToolChoiceAnyDict, ToolChoiceToolDict] module-attribute ¶

logger = logging.getLogger(__name__) module-attribute ¶

ContentBlock ¶

ContextWindowOverflowException ¶

LlamaCppModel ¶

Audio analysis¶

Image analysis¶

LlamaCppConfig ¶

__init__(base_url='http://localhost:8080', timeout=None, **model_config) ¶

get_config() ¶

stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, **kwargs) async ¶

structured_output(output_model, prompt, system_prompt=None, **kwargs) async ¶

update_config(**model_config) ¶

Model ¶

get_config() abstractmethod ¶

stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, system_prompt_content=None, **kwargs) abstractmethod ¶

structured_output(output_model, prompt, system_prompt=None, **kwargs) abstractmethod ¶

update_config(**model_config) abstractmethod ¶

ModelThrottledException ¶

__init__(message) ¶

StreamEvent ¶

ToolSpec ¶

validate_config_keys(config_dict, config_class) ¶

warn_on_tool_choice_not_supported(tool_choice) ¶

`strands.models.llamacpp` ¶

`Messages = List[Message]` `module-attribute` ¶

`T = TypeVar('T', bound=BaseModel)` `module-attribute` ¶

`ToolChoice = Union[ToolChoiceAutoDict, ToolChoiceAnyDict, ToolChoiceToolDict]` `module-attribute` ¶

`logger = logging.getLogger(name)` `module-attribute` ¶

`ContentBlock` ¶

`ContextWindowOverflowException` ¶

`LlamaCppModel` ¶

`LlamaCppConfig` ¶

`init(base_url='http://localhost:8080', timeout=None, **model_config)` ¶

`get_config()` ¶

`stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, **kwargs)` `async` ¶

`structured_output(output_model, prompt, system_prompt=None, **kwargs)` `async` ¶

`update_config(**model_config)` ¶

`Model` ¶

`get_config()` `abstractmethod` ¶

`stream(messages, tool_specs=None, system_prompt=None, *, tool_choice=None, system_prompt_content=None, **kwargs)` `abstractmethod` ¶

`structured_output(output_model, prompt, system_prompt=None, **kwargs)` `abstractmethod` ¶

`update_config(**model_config)` `abstractmethod` ¶

`ModelThrottledException` ¶

`init(message)` ¶

`StreamEvent` ¶

`ToolSpec` ¶

`validate_config_keys(config_dict, config_class)` ¶

`warn_on_tool_choice_not_supported(tool_choice)` ¶