Add a ToolLike protocol to decouple tool schema serialization boundary

Author: anbuzin

examples/temporal-durable/workflow.py

@@ -28,12 +28,12 @@ def __init__(
     async def stream(
         self,
         messages: list[ai.Message],
-        tools: Sequence[ai.ToolSchema] | None = None,
+        tools: Sequence[ai.ToolLike] | None = None,
     ) -> AsyncGenerator[ai.Message, None]:
         result = await self.call_fn(
             activities.LLMCallParams(
                 messages=[m.model_dump() for m in messages],
-                tool_schemas=[t.model_dump() for t in (tools or [])],
+                tool_schemas=[t.schema.model_dump() for t in (tools or [])],
             )
         )
         yield ai.Message.model_validate(result.message)

src/vercel_ai_sdk/__init__.py

@@ -12,7 +12,7 @@
     HookPart,
     make_messages,
 )
-from .core.tools import ToolSchema, Tool, tool
+from .core.tools import ToolLike, ToolSchema, Tool, tool
 from .core.llm import LanguageModel
 from .core.streams import StreamResult, stream
 from .core.runtime import (
@@ -37,6 +37,7 @@
     "ToolPart",
     "ToolDelta",
     "ReasoningPart",
+    "ToolLike",
     "ToolSchema",
     "Tool",
     "LanguageModel",

src/vercel_ai_sdk/anthropic/__init__.py

@@ -10,7 +10,7 @@
 from .. import core
 
 
-def _tools_to_anthropic(tools: Sequence[core.tools.Tool]) -> list[dict[str, Any]]:
+def _tools_to_anthropic(tools: Sequence[core.tools.ToolLike]) -> list[dict[str, Any]]:
     """Convert internal Tool objects to Anthropic tool schema format."""
     return [
         {
@@ -117,7 +117,7 @@ def __init__(
     async def stream_events(
         self,
         messages: list[core.messages.Message],
-        tools: Sequence[core.tools.Tool] | None = None,
+        tools: Sequence[core.tools.ToolLike] | None = None,
     ) -> AsyncGenerator[core.llm.StreamEvent, None]:
         """Yield raw stream events from Anthropic API."""
         system_prompt, anthropic_messages = _messages_to_anthropic(messages)
@@ -206,7 +206,7 @@ async def stream_events(
     async def stream(
         self,
         messages: list[core.messages.Message],
-        tools: Sequence[core.tools.Tool] | None = None,
+        tools: Sequence[core.tools.ToolLike] | None = None,
     ) -> AsyncGenerator[core.messages.Message, None]:
         """Stream Messages (uses StreamProcessor internally)."""
         handler = core.llm.StreamHandler()

src/vercel_ai_sdk/core/llm.py

@@ -215,15 +215,15 @@ class LanguageModel(abc.ABC):
     async def stream(
         self,
         messages: list[messages_.Message],
-        tools: Sequence[tools_.Tool] | None = None,
+        tools: Sequence[tools_.ToolLike] | None = None,
     ) -> AsyncGenerator[messages_.Message, None]:
         raise NotImplementedError
         yield
 
     async def buffer(
         self,
         messages: list[messages_.Message],
-        tools: Sequence[tools_.Tool] | None = None,
+        tools: Sequence[tools_.ToolLike] | None = None,
     ) -> messages_.Message:
         """Drain the stream and return the final message."""
         final = None

src/vercel_ai_sdk/core/runtime.py

@@ -187,7 +187,7 @@ def _find_runtime_param(fn: Callable[..., Any]) -> str | None:
 async def stream_step(
     llm: llm_.LanguageModel,
     messages: list[messages_.Message],
-    tools: Sequence[tools_.Tool] | None = None,
+    tools: Sequence[tools_.ToolLike] | None = None,
     label: str | None = None,
 ) -> AsyncGenerator[messages_.Message, None]:
     """Single LLM call that streams to Runtime."""
@@ -243,7 +243,7 @@ async def execute_tool(
 async def stream_loop(
     llm: llm_.LanguageModel,
     messages: list[messages_.Message],
-    tools: Sequence[tools_.Tool],
+    tools: Sequence[tools_.ToolLike],
     label: str | None = None,
 ) -> streams_.StreamResult:
     """Agent loop: stream LLM, execute tools, repeat until done."""

src/vercel_ai_sdk/core/tools.py

@@ -3,7 +3,7 @@
 import inspect
 import json
 from collections.abc import Awaitable, Callable
-from typing import TYPE_CHECKING, Any, get_type_hints
+from typing import TYPE_CHECKING, Any, Protocol, get_type_hints, runtime_checkable
 
 import pydantic
 
@@ -27,6 +27,18 @@ def _is_runtime_type(hint: Any) -> bool:
     return hint is Runtime
 
 
+@runtime_checkable
+class ToolLike(Protocol):
+    """Anything the LLM layer can use as a tool definition."""
+
+    @property
+    def name(self) -> str: ...
+    @property
+    def description(self) -> str: ...
+    @property
+    def param_schema(self) -> dict[str, Any]: ...
+
+
 class ToolSchema(pydantic.BaseModel):
     """What the LLM sees: name, description, and JSON Schema for parameters."""
 

src/vercel_ai_sdk/openai/__init__.py

@@ -9,7 +9,7 @@
 from .. import core
 
 
-def _tools_to_openai(tools: Sequence[core.tools.Tool]) -> list[dict[str, Any]]:
+def _tools_to_openai(tools: Sequence[core.tools.ToolLike]) -> list[dict[str, Any]]:
     """Convert internal Tool objects to OpenAI tool schema format."""
     return [
         {
@@ -132,7 +132,7 @@ def __init__(
     async def stream_events(
         self,
         messages: list[core.messages.Message],
-        tools: Sequence[core.tools.Tool] | None = None,
+        tools: Sequence[core.tools.ToolLike] | None = None,
     ) -> AsyncGenerator[core.llm.StreamEvent, None]:
         """Yield raw stream events from OpenAI API."""
         openai_messages = _messages_to_openai(messages)
@@ -243,7 +243,7 @@ async def stream_events(
     async def stream(
         self,
         messages: list[core.messages.Message],
-        tools: Sequence[core.tools.Tool] | None = None,
+        tools: Sequence[core.tools.ToolLike] | None = None,
     ) -> AsyncGenerator[core.messages.Message, None]:
         """Stream Messages (uses StreamHandler internally)."""
         handler = core.llm.StreamHandler()

tests/conftest.py

@@ -17,7 +17,7 @@ def __init__(self, responses: list[list[messages.Message]]) -> None:
     async def stream(
         self,
         messages: list[messages.Message],
-        tools: Sequence[ai.Tool] | None = None,
+        tools: Sequence[ai.ToolLike] | None = None,
     ) -> AsyncGenerator[messages.Message, None]:
         if self._call_index >= len(self._responses):
             raise RuntimeError("MockLLM: no more responses configured")