Implementing unified format for function calling.

src/pipecat/adapters/base_llm_adapter.py

@@ -0,0 +1,22 @@
+from abc import ABC, abstractmethod
+from typing import Any, List, Union, cast
+
+from loguru import logger
+
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+
+
+class BaseLLMAdapter(ABC):
+    @abstractmethod
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Any]:
+        """Converts tools to the provider's format."""
+        pass
+
+    def from_standard_tools(self, tools: Any) -> List[Any]:
+        if isinstance(tools, ToolsSchema):
+            logger.debug(f"Retrieving the tools using the adapter: {type(self)}")
+            return self.to_provider_tools_format(tools)
+        # Fallback to return the same tools in case they are not in a standard format
+        return tools
+
+    # TODO: we can move the logic to also handle the Messages here

src/pipecat/adapters/schemas/function_schema.py

@@ -0,0 +1,55 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Any, Dict, List
+
+
+class FunctionSchema:
+    def __init__(
+        self, name: str, description: str, properties: Dict[str, Any], required: List[str]
+    ) -> None:
+        """Standardized function schema representation.
+
+        :param name: Name of the function.
+        :param description: Description of the function.
+        :param properties: Dictionary defining properties types and descriptions.
+        :param required: List of required parameters.
+        """
+        self._name = name
+        self._description = description
+        self._properties = properties
+        self._required = required
+
+    def to_default_dict(self) -> Dict[str, Any]:
+        """Converts the function schema to a dictionary.
+
+        :return: Dictionary representation of the function schema.
+        """
+        return {
+            "name": self._name,
+            "description": self._description,
+            "parameters": {
+                "type": "object",
+                "properties": self._properties,
+                "required": self._required,
+            },
+        }
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def description(self) -> str:
+        return self._description
+
+    @property
+    def properties(self) -> Dict[str, Any]:
+        return self._properties
+
+    @property
+    def required(self) -> List[str]:
+        return self._required

src/pipecat/adapters/schemas/tools_schema.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from enum import Enum
+from typing import Any, Dict, List
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+
+
+class AdapterType(Enum):
+    GEMINI = "gemini"  # that is the only service where we are able to add custom tools for now
+
+
+class ToolsSchema:
+    def __init__(
+        self,
+        standard_tools: List[FunctionSchema],
+        custom_tools: Dict[AdapterType, List[Dict[str, Any]]] = None,
+    ) -> None:
+        """
+        A schema for tools that includes both standardized function schemas
+        and custom tools that do not follow the FunctionSchema format.
+
+        :param standard_tools: List of tools following FunctionSchema.
+        :param custom_tools: List of tools in a custom format (e.g., search_tool).
+        """
+        self._standard_tools = standard_tools
+        self._custom_tools = custom_tools
+
+    @property
+    def standard_tools(self) -> List[FunctionSchema]:
+        return self._standard_tools
+
+    @property
+    def custom_tools(self) -> Dict[AdapterType, List[Dict[str, Any]]]:
+        return self._custom_tools
+
+    @custom_tools.setter
+    def custom_tools(self, value: Dict[AdapterType, List[Dict[str, Any]]]) -> None:
+        self._custom_tools = value

src/pipecat/adapters/services/anthropic_adapter.py

@@ -0,0 +1,34 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Any, Dict, List, Union
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+
+
+class AnthropicLLMAdapter(BaseLLMAdapter):
+    @staticmethod
+    def _to_anthropic_function_format(function: FunctionSchema) -> Dict[str, Any]:
+        return {
+            "name": function.name,
+            "description": function.description,
+            "input_schema": {
+                "type": "object",
+                "properties": function.properties,
+                "required": function.required,
+            },
+        }
+
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
+        """Converts function schemas to Anthropic's function-calling format.
+
+        :return: Anthropic formatted function call definition.
+        """
+
+        functions_schema = tools_schema.standard_tools
+        return [self._to_anthropic_function_format(func) for func in functions_schema]

src/pipecat/adapters/services/gemini_adapter.py

@@ -0,0 +1,28 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Any, Dict, List, Union
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+
+
+class GeminiLLMAdapter(BaseLLMAdapter):
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
+        """Converts function schemas to Gemini's function-calling format.
+
+        :return: Gemini formatted function call definition.
+        """
+
+        functions_schema = tools_schema.standard_tools
+        formatted_standard_tools = [
+            {"function_declarations": [func.to_default_dict() for func in functions_schema]}
+        ]
+        custom_gemini_tools = []
+        if tools_schema.custom_tools:
+            custom_gemini_tools = tools_schema.custom_tools.get(AdapterType.GEMINI, [])
+
+        return formatted_standard_tools + custom_gemini_tools

src/pipecat/adapters/services/open_ai_adapter.py

@@ -0,0 +1,24 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+from typing import List
+
+from openai.types.chat import ChatCompletionToolParam
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+
+
+class OpenAILLMAdapter(BaseLLMAdapter):
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[ChatCompletionToolParam]:
+        """Converts function schemas to OpenAI's function-calling format.
+
+        :return: OpenAI formatted function call definition.
+        """
+        functions_schema = tools_schema.standard_tools
+        return [
+            ChatCompletionToolParam(type="function", function=func.to_default_dict())
+            for func in functions_schema
+        ]

src/pipecat/adapters/services/open_ai_realtime_adapter.py

@@ -0,0 +1,34 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+from typing import Any, Dict, List, Union
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+
+
+class OpenAIRealtimeLLMAdapter(BaseLLMAdapter):
+    @staticmethod
+    def _to_openai_realtime_function_format(function: FunctionSchema) -> Dict[str, Any]:
+        return {
+            "type": "function",
+            "name": function.name,
+            "description": function.description,
+            "parameters": {
+                "type": "object",
+                "properties": function.properties,
+                "required": function.required,
+            },
+        }
+
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
+        """Converts function schemas to Openai Realtime function-calling format.
+
+        :return: Openai Realtime formatted function call definition.
+        """
+
+        functions_schema = tools_schema.standard_tools
+        return [self._to_openai_realtime_function_format(func) for func in functions_schema]

src/pipecat/processors/aggregators/openai_llm_context.py

@@ -20,6 +20,8 @@ from openai.types.chat import (
 )
 from PIL import Image
 
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.frames.frames import (
     AudioRawFrame,
     Frame,
@@ -44,13 +46,20 @@ class OpenAILLMContext:
     def __init__(
         self,
         messages: Optional[List[ChatCompletionMessageParam]] = None,
-        tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN,
+        tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = NOT_GIVEN,
         tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = NOT_GIVEN,
     ):
         self._messages: List[ChatCompletionMessageParam] = messages if messages else []
         self._tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven = tool_choice
-        self._tools: List[ChatCompletionToolParam] | NotGiven = tools
+        self._tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = tools
         self._user_image_request_context = {}
+        self._llm_adapter: Optional[BaseLLMAdapter] = None
+
+    def get_llm_adapter(self) -> Optional[BaseLLMAdapter]:
+        return self._llm_adapter
+
+    def set_llm_adapter(self, llm_adapter: BaseLLMAdapter):
+        self._llm_adapter = llm_adapter
 
     @staticmethod
     def from_messages(messages: List[dict]) -> "OpenAILLMContext":
@@ -67,7 +76,9 @@ class OpenAILLMContext:
         return self._messages
 
     @property
-    def tools(self) -> List[ChatCompletionToolParam] | NotGiven:
+    def tools(self) -> List[ChatCompletionToolParam] | NotGiven | List[Any]:
+        if self._llm_adapter:
+            return self._llm_adapter.from_standard_tools(self._tools)
         return self._tools
 
     @property
@@ -152,7 +163,7 @@ class OpenAILLMContext:
     def set_tool_choice(self, tool_choice: ChatCompletionToolChoiceOptionParam | NotGiven):
         self._tool_choice = tool_choice
 
-    def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven = NOT_GIVEN):
+    def set_tools(self, tools: List[ChatCompletionToolParam] | NotGiven | ToolsSchema = NOT_GIVEN):
         if tools != NOT_GIVEN and len(tools) == 0:
             tools = NOT_GIVEN
         self._tools = tools

src/pipecat/services/ai_services.py

@@ -8,10 +8,12 @@ import asyncio
 import io
 import wave
 from abc import abstractmethod
-from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Tuple
+from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Tuple, Type
 
 from loguru import logger
 
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
 from pipecat.audio.utils import calculate_audio_volume, exp_smoothing
 from pipecat.frames.frames import (
     AudioRawFrame,
@@ -137,10 +139,23 @@ class AIService(FrameProcessor):
 class LLMService(AIService):
     """This class is a no-op but serves as a base class for LLM services."""
 
+    # OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations.
+    # However, subclasses should override this with a more specific adapter when necessary.
+    adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self._callbacks = {}
         self._start_callbacks = {}
+        self._adapter = self.adapter_class()
+
+    def get_llm_adapter(self) -> BaseLLMAdapter:
+        return self._adapter
+
+    def create_context_aggregator(
+        self, context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True
+    ) -> Any:
+        pass
 
         self._register_event_handler("on_completion_timeout")
 

src/pipecat/services/anthropic.py

@@ -18,6 +18,7 @@ from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
 
+from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
 from pipecat.frames.frames import (
     Frame,
     FunctionCallInProgressFrame,
@@ -85,6 +86,9 @@ class AnthropicLLMService(LLMService):
     use `AsyncAnthropicBedrock` and `AsyncAnthropicVertex` clients
     """
 
+    # Overriding the default adapter to use the Anthropic one.
+    adapter_class = AnthropicLLMAdapter
+
     class InputParams(BaseModel):
         enable_prompt_caching_beta: Optional[bool] = False
         max_tokens: Optional[int] = Field(default_factory=lambda: 4096, ge=1)
@@ -123,8 +127,8 @@ class AnthropicLLMService(LLMService):
     def enable_prompt_caching_beta(self) -> bool:
         return self._enable_prompt_caching_beta
 
-    @staticmethod
     def create_context_aggregator(
+        self,
         context: OpenAILLMContext,
         *,
         user_kwargs: Mapping[str, Any] = {},
@@ -149,6 +153,8 @@ class AnthropicLLMService(LLMService):
             AnthropicContextAggregatorPair.
 
         """
+        context.set_llm_adapter(self.get_llm_adapter())
+
         if isinstance(context, OpenAILLMContext):
             context = AnthropicLLMContext.from_openai_context(context)
         user = AnthropicUserContextAggregator(context, **user_kwargs)
@@ -382,6 +388,7 @@ class AnthropicLLMContext(OpenAILLMContext):
             tools=openai_context.tools,
             tool_choice=openai_context.tool_choice,
         )
+        self.set_llm_adapter(openai_context.get_llm_adapter())
         self._restructure_from_openai_messages()
         return self
 

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -9,12 +9,14 @@ import base64
 import json
 from dataclasses import dataclass
 from enum import Enum
-from typing import Any, Dict, List, Mapping, Optional
+from typing import Any, Dict, List, Mapping, Optional, Union
 
 import websockets
 from loguru import logger
 from pydantic import BaseModel, Field
 
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
@@ -152,6 +154,9 @@ class InputParams(BaseModel):
 
 
 class GeminiMultimodalLiveLLMService(LLMService):
+    # Overriding the default adapter to use the Gemini one.
+    adapter_class = GeminiLLMAdapter
+
     def __init__(
         self,
         *,
@@ -162,7 +167,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         start_audio_paused: bool = False,
         start_video_paused: bool = False,
         system_instruction: Optional[str] = None,
-        tools: Optional[List[dict]] = None,
+        tools: Optional[Union[List[dict], ToolsSchema]] = None,
         transcribe_user_audio: bool = False,
         transcribe_model_audio: bool = False,
         params: InputParams = InputParams(),
@@ -435,7 +440,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 )
             if self._tools:
                 logger.debug(f"Gemini is configuring to use tools{self._tools}")
-                config.setup.tools = self._tools
+                config.setup.tools = self.get_llm_adapter().from_standard_tools(self._tools)
             await self.send_client_event(config)
 
         except Exception as e:
@@ -726,6 +731,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
             encapsulated in an GeminiMultimodalLiveContextAggregatorPair.
 
         """
+        context.set_llm_adapter(self.get_llm_adapter())
+
         GeminiMultimodalLiveContext.upgrade(context)
         user = GeminiMultimodalLiveUserContextAggregator(context, **user_kwargs)
 

src/pipecat/services/google/google.py

@@ -15,6 +15,8 @@ from google.api_core.exceptions import DeadlineExceeded
 from openai import AsyncStream
 from openai.types.chat import ChatCompletionChunk
 
+from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
+
 # Suppress gRPC fork warnings
 os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
 
@@ -950,6 +952,9 @@ class GoogleLLMService(LLMService):
     franca for all LLM services, so that it is easy to switch between different LLMs.
     """
 
+    # Overriding the default adapter to use the Gemini one.
+    adapter_class = GeminiLLMAdapter
+
     class InputParams(BaseModel):
         max_tokens: Optional[int] = Field(default=4096, ge=1)
         temperature: Optional[float] = Field(default=None, ge=0.0, le=2.0)
@@ -1180,8 +1185,8 @@ class GoogleLLMService(LLMService):
         if context:
             await self._process_context(context)
 
-    @staticmethod
     def create_context_aggregator(
+        self,
         context: OpenAILLMContext,
         *,
         user_kwargs: Mapping[str, Any] = {},
@@ -1206,6 +1211,8 @@ class GoogleLLMService(LLMService):
             GoogleContextAggregatorPair.
 
         """
+        context.set_llm_adapter(self.get_llm_adapter())
+
         if isinstance(context, OpenAILLMContext):
             context = GoogleLLMContext.upgrade_to_google(context)
         user = GoogleUserContextAggregator(context, **user_kwargs)

src/pipecat/services/grok.py

@@ -206,8 +206,8 @@ class GrokLLMService(OpenAILLMService):
         if tokens.completion_tokens > self._completion_tokens:
             self._completion_tokens = tokens.completion_tokens
 
-    @staticmethod
     def create_context_aggregator(
+        self,
         context: OpenAILLMContext,
         *,
         user_kwargs: Mapping[str, Any] = {},
@@ -232,6 +232,8 @@ class GrokLLMService(OpenAILLMService):
             GrokContextAggregatorPair.
 
         """
+        context.set_llm_adapter(self.get_llm_adapter())
+
         user = OpenAIUserContextAggregator(context, **user_kwargs)
         assistant = GrokAssistantContextAggregator(context, **assistant_kwargs)
         return GrokContextAggregatorPair(_user=user, _assistant=assistant)

src/pipecat/services/openai.py

@@ -343,8 +343,8 @@ class OpenAILLMService(BaseOpenAILLMService):
     ):
         super().__init__(model=model, params=params, **kwargs)
 
-    @staticmethod
     def create_context_aggregator(
+        self,
         context: OpenAILLMContext,
         *,
         user_kwargs: Mapping[str, Any] = {},
@@ -369,6 +369,7 @@ class OpenAILLMService(BaseOpenAILLMService):
             OpenAIContextAggregatorPair.
 
         """
+        context.set_llm_adapter(self.get_llm_adapter())
         user = OpenAIUserContextAggregator(context, **user_kwargs)
         assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs)
         return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)

src/pipecat/services/openai_realtime_beta/openai.py

@@ -12,6 +12,8 @@ from typing import Any, Mapping
 
 from loguru import logger
 
+from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
+
 try:
     import websockets
 except ModuleNotFoundError as e:
@@ -76,6 +78,9 @@ class OpenAIUnhandledFunctionException(Exception):
 
 
 class OpenAIRealtimeBetaLLMService(LLMService):
+    # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one.
+    adapter_class = OpenAIRealtimeLLMAdapter
+
     def __init__(
         self,
         *,
@@ -596,6 +601,8 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             OpenAIContextAggregatorPair.
 
         """
+        context.set_llm_adapter(self.get_llm_adapter())
+
         OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
         user = OpenAIRealtimeUserContextAggregator(context, **user_kwargs)