feat: add host model provider interface in wit contract, host model provider to TS and SageMaker support to PY to enable host model calling for wasm architecture

strands-py/strands/_wasm_host.py

@@ -82,6 +82,22 @@ def log(self, level: str, message: str, context: typing.Optional[str]) -> None:
         raise NotImplementedError
 
 
+class ModelProviderBase(ABC):
+    @abstractmethod
+    def invoke(
+        self,
+        messages: str,
+        system_prompt: typing.Optional[str],
+        tool_specs: typing.Optional[str],
+        config: str,
+    ) -> list[str]:
+        """Invoke the host-side model provider.
+
+        Returns a list of model event JSON strings.
+        """
+        raise NotImplementedError
+
+
 def _run_sync(coro: typing.Coroutine) -> typing.Any:
     """Run an async coroutine from sync context, even if an event loop is running."""
     try:
@@ -212,6 +228,9 @@ def _build_model_config_variant(cfg: ModelConfigInput) -> Variant:
             }
         )
         return Variant("gemini", payload)
+    if provider == "host-model":
+        payload = _rec(config=cfg.additional_config or "{}")
+        return Variant("host-model", payload)
     raise ValueError(f"unknown model provider: {provider}")
 
 
@@ -476,6 +495,47 @@ def log_fn(store_ctx: typing.Any, entry: typing.Any) -> None:
     return log_fn
 
 
+def _make_model_invoke_fn(
+    provider: ModelProviderBase | None,
+) -> typing.Callable[..., typing.Any]:
+    """Create the callback for the model-provider.invoke WIT import."""
+
+    async def invoke_fn(store_ctx: typing.Any, args: typing.Any) -> Variant:
+        if provider is None:
+            return Variant("err", "no host model provider configured")
+        messages = getattr(args, "messages")
+        system_prompt = getattr(args, "system-prompt")
+        config = getattr(args, "config")
+
+        # Serialize tool specs if present
+        raw_specs = getattr(args, "tool-specs")
+        tool_specs_json: str | None = None
+        if raw_specs is not None:
+            import json as _json
+
+            tool_specs_json = _json.dumps(
+                [
+                    {
+                        "name": getattr(s, "name"),
+                        "description": getattr(s, "description"),
+                        "inputSchema": getattr(s, "input-schema"),
+                    }
+                    for s in raw_specs
+                ]
+            )
+
+        try:
+            event_jsons = provider.invoke(messages, system_prompt, tool_specs_json, config)
+            log.debug("model-provider invoke returned %d events", len(event_jsons))
+            import json as _json2
+            return Variant("ok", _json2.dumps(event_jsons))
+        except Exception as exc:
+            log.error("model-provider invoke failed: %s", exc)
+            return Variant("err", str(exc))
+
+    return invoke_fn
+
+
 # ---------------------------------------------------------------------------
 # WasmAgent — drop-in replacement for the former native Agent class
 # ---------------------------------------------------------------------------
@@ -491,6 +551,7 @@ def __init__(
         tools: list[ToolSpec] | None,
         tool_dispatcher: ToolDispatcherBase | None,
         log_handler: LogHandlerBase | None,
+        model_provider: ModelProviderBase | None = None,
         use_callback_relay: bool = False,
     ):
         engine, component = _get_engine_and_component()
@@ -506,6 +567,8 @@ def __init__(
                 tp.add_func("call-tools", _make_call_tools_fn(tool_dispatcher))
             with root.add_instance("strands:agent/host-log") as hl:
                 hl.add_func("log", _make_log_fn(log_handler))
+            with root.add_instance("strands:agent/model-provider") as mp:
+                mp.add_func_async("invoke", _make_model_invoke_fn(model_provider))
 
         # --- store ---
         store = Store(engine)
@@ -606,4 +669,4 @@ def get_messages(self) -> str:
 
     def set_messages(self, json: str) -> None:
         """Sync wrapper — safe from any context (inside or outside event loop)."""
-        _run_sync(self.set_messages_async(json))
+        _run_sync(self.set_messages_async(json))

strands-py/strands/agent/__init__.py

@@ -279,7 +279,20 @@ def __init__(
             sp_blocks = json.dumps(system_prompt)
             sp_str = None
 
-        model_config = self._build_model_config(resolve_model(model))
+        # Detect host-side model providers (have stream() method — e.g. SageMaker, Mistral, Writer).
+        # These run on the Python side and are invoked by the WASM guest via model-provider import.
+        model_provider_callback = None
+        if model is not None and hasattr(model, "stream") and callable(getattr(model, "stream", None)):
+            from strands.models.host_adapter import HostModelAdapter
+
+            model_provider_callback = HostModelAdapter(model)
+            model_config = _ModelConfigInput(
+                provider="host-model",
+                additional_config=json.dumps({"provider_type": type(model).__name__}),
+            )
+        else:
+            model_config = self._build_model_config(resolve_model(model))
+
         tool_specs = (
             [
                 _ToolSpec(
@@ -300,6 +313,7 @@ def __init__(
             tools=tool_specs,
             tool_dispatcher=self._dispatcher,
             log_handler=_LogHandler(),
+            model_provider=model_provider_callback,
             use_callback_relay=False,
         )
 
@@ -766,4 +780,4 @@ def cleanup(self) -> None:
 # Re-export for test compatibility
 from strands.agent.conversation_manager import NullConversationManager  # noqa: E402
 
-__all__ = ["Agent", "AgentResult", "NullConversationManager"]
+__all__ = ["Agent", "AgentResult", "NullConversationManager"]

strands-py/strands/models/__init__.py

@@ -3,5 +3,6 @@
 from strands.models.gemini import GeminiModel
 from strands.models.model import Model
 from strands.models.openai import OpenAIModel
+from strands.models.sagemaker import SageMakerAIModel
 
-__all__ = ["AnthropicModel", "BedrockModel", "GeminiModel", "Model", "OpenAIModel"]
+__all__ = ["AnthropicModel", "BedrockModel", "GeminiModel", "Model", "OpenAIModel", "SageMakerAIModel"]

strands-py/strands/models/_validation.py

@@ -0,0 +1,64 @@
+"""Configuration validation utilities for model providers."""
+
+import warnings
+from collections.abc import Mapping
+from typing import Any
+
+from typing_extensions import get_type_hints
+
+from ..types.content import ContentBlock
+from ..types.tools import ToolChoice
+
+
+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,
+        )
+
+
+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,
+        )
+
+
+def _has_location_source(content: ContentBlock) -> bool:
+    """Check if a content block contains a location source.
+
+    Providers need to explicitly define an implementation to support content locations.
+
+    Args:
+        content: Content block to check.
+
+    Returns:
+        True if the content block contains an location source, False otherwise.
+    """
+    if "image" in content:
+        return "location" in content["image"].get("source", {})
+    if "document" in content:
+        return "location" in content["document"].get("source", {})
+    if "video" in content:
+        return "location" in content["video"].get("source", {})
+    return False