Merge pull request #1211 from pipecat-ai/function_calling_unified_format

Unified format for function calling

CHANGELOG.md

@@ -9,6 +9,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added support for a unified format for specifying function calling across all LLM services.
+  ```python
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location"],
+    )
+    tools = ToolsSchema(standard_tools=[weather_function])
+  ```
+
 - Added `speech_threshold` parameter to `GladiaSTTService`.
 
 - Allow passing user (`user_kwargs`) and assistant (`assistant_kwargs`) context

examples/unified-format-function-calling/base_function_calling.py

@@ -0,0 +1,134 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+import sys
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import TTSSpeakFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.ai_services import LLMService
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+load_dotenv(override=True)
+
+
+async def start_fetch_weather(function_name, llm, context):
+    """Push a frame to the LLM; this is handy when the LLM response might take a while."""
+    await llm.push_frame(TTSSpeakFrame("Let me check on that."))
+    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
+
+
+async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
+    await result_callback({"conditions": "nice", "temperature": "75"})
+
+
+class WeatherBot:
+    """Generic base class for setting up and running an LLM-powered bot."""
+
+    def __init__(self, llm: LLMService):
+        """Initialize the base handler with a specific LLM."""
+        self.llm = llm
+
+    async def run(self):
+        """Set up and start the processing pipeline."""
+        async with aiohttp.ClientSession() as session:
+            (room_url, token) = await configure(session)
+
+            transport = DailyTransport(
+                room_url,
+                token,
+                "Respond bot",
+                DailyParams(
+                    audio_out_enabled=True,
+                    transcription_enabled=True,
+                    vad_enabled=True,
+                    vad_analyzer=SileroVADAnalyzer(),
+                ),
+            )
+
+            tts = CartesiaTTSService(
+                api_key=os.getenv("CARTESIA_API_KEY"),
+                voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+            )
+
+            # Register a function_name of None to get all functions
+            # sent to the same callback with an additional function_name parameter.
+            self.llm.register_function(
+                None, fetch_weather_from_api, start_callback=start_fetch_weather
+            )
+
+            weather_function = FunctionSchema(
+                name="get_current_weather",
+                description="Get the current weather",
+                properties={
+                    "location": {
+                        "type": "string",
+                        "description": "The city and state, e.g. San Francisco, CA",
+                    },
+                    "format": {
+                        "type": "string",
+                        "enum": ["celsius", "fahrenheit"],
+                        "description": "The temperature unit to use. Infer this from the user's location.",
+                    },
+                },
+                required=["location"],
+            )
+            tools = ToolsSchema(standard_tools=[weather_function])
+
+            messages = [
+                {
+                    "role": "system",
+                    "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation.",
+                },
+                {"role": "user", "content": " Start the conversation by introducing yourself."},
+            ]
+
+            context = OpenAILLMContext(messages, tools)
+            context_aggregator = self.llm.create_context_aggregator(context)
+
+            pipeline = Pipeline(
+                [
+                    transport.input(),
+                    context_aggregator.user(),
+                    self.llm,
+                    tts,
+                    transport.output(),
+                    context_aggregator.assistant(),
+                ]
+            )
+
+            task = PipelineTask(
+                pipeline,
+                params=PipelineParams(
+                    allow_interruptions=True,
+                    enable_metrics=True,
+                    enable_usage_metrics=True,
+                    report_only_initial_ttfb=True,
+                ),
+            )
+
+            @transport.event_handler("on_first_participant_joined")
+            async def on_first_participant_joined(transport, participant):
+                await transport.capture_participant_transcription(participant["id"])
+                await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+            runner = PipelineRunner()
+            await runner.run(task)

examples/unified-format-function-calling/multimodal_base_function_calling.py

@@ -0,0 +1,126 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import sys
+from typing import List
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+from runner import configure
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import TTSSpeakFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.ai_services import LLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+load_dotenv(override=True)
+
+
+async def start_fetch_weather(function_name, llm, context):
+    """Push a frame to the LLM; this is handy when the LLM response might take a while."""
+    await llm.push_frame(TTSSpeakFrame("Let me check on that."))
+    logger.debug(f"Starting fetch_weather_from_api with function_name: {function_name}")
+
+
+async def fetch_weather_from_api(function_name, tool_call_id, args, llm, context, result_callback):
+    await result_callback({"conditions": "nice", "temperature": "75"})
+
+
+class MultimodalWeatherBot:
+    """Generic base class for setting up and running an LLM-powered bot."""
+
+    def __init__(self, llm: LLMService):
+        """Initialize the base handler with a specific LLM."""
+        self.llm = llm
+
+    @staticmethod
+    def tools() -> ToolsSchema:
+        weather_function = FunctionSchema(
+            name="get_current_weather",
+            description="Get the current weather",
+            properties={
+                "location": {
+                    "type": "string",
+                    "description": "The city and state, e.g. San Francisco, CA",
+                },
+                "format": {
+                    "type": "string",
+                    "enum": ["celsius", "fahrenheit"],
+                    "description": "The temperature unit to use. Infer this from the user's location.",
+                },
+            },
+            required=["location"],
+        )
+        return ToolsSchema(standard_tools=[weather_function])
+
+    async def run(self):
+        """Set up and start the processing pipeline."""
+        async with aiohttp.ClientSession() as session:
+            (room_url, token) = await configure(session)
+
+            transport = DailyTransport(
+                room_url,
+                token,
+                "Respond bot",
+                DailyParams(
+                    audio_out_enabled=True,
+                    vad_enabled=True,
+                    vad_analyzer=SileroVADAnalyzer(),
+                    vad_audio_passthrough=True,
+                ),
+            )
+
+            # Register a function_name of None to get all functions
+            # sent to the same callback with an additional function_name parameter.
+            self.llm.register_function(
+                None, fetch_weather_from_api, start_callback=start_fetch_weather
+            )
+
+            messages = [
+                {
+                    "role": "system",
+                    "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation.",
+                },
+                {"role": "user", "content": " Start the conversation by introducing yourself."},
+            ]
+
+            context = OpenAILLMContext(messages, MultimodalWeatherBot.tools())
+            context_aggregator = self.llm.create_context_aggregator(context)
+
+            pipeline = Pipeline(
+                [
+                    transport.input(),
+                    context_aggregator.user(),
+                    self.llm,
+                    transport.output(),
+                    context_aggregator.assistant(),
+                ]
+            )
+
+            task = PipelineTask(
+                pipeline,
+                params=PipelineParams(
+                    allow_interruptions=True,
+                ),
+            )
+
+            @transport.event_handler("on_first_participant_joined")
+            async def on_first_participant_joined(transport, participant):
+                await transport.capture_participant_transcription(participant["id"])
+                await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+            runner = PipelineRunner()
+            await runner.run(task)

examples/unified-format-function-calling/runner.py

@@ -0,0 +1,64 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+from typing import Optional
+
+import aiohttp
+
+from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
+
+
+async def configure(aiohttp_session: aiohttp.ClientSession):
+    (url, token, _) = await configure_with_args(aiohttp_session)
+    return (url, token)
+
+
+async def configure_with_args(
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
+):
+    if not parser:
+        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    if not key:
+        raise Exception(
+            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
+        )
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token, args)

examples/unified-format-function-calling/standard-function-calling-anthropic.py

@@ -0,0 +1,29 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.anthropic import AnthropicLLMService
+
+load_dotenv(override=True)
+
+
+class AnthropicWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = AnthropicLLMService(
+            api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620"
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(AnthropicWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-azure.py

@@ -0,0 +1,31 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.azure import AzureLLMService
+
+load_dotenv(override=True)
+
+
+class AzureWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = AzureLLMService(
+            api_key=os.getenv("AZURE_CHATGPT_API_KEY"),
+            endpoint=os.getenv("AZURE_CHATGPT_ENDPOINT"),
+            model=os.getenv("AZURE_CHATGPT_MODEL"),
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(AzureWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-cerebras.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.cerebras import CerebrasLLMService
+
+load_dotenv(override=True)
+
+
+class CerebrasWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = CerebrasLLMService(api_key=os.getenv("CEREBRAS_API_KEY"), model="llama-3.3-70b")
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(CerebrasWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-deepseek.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.deepseek import DeepSeekLLMService
+
+load_dotenv(override=True)
+
+
+class DeepSeekWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = DeepSeekLLMService(api_key=os.getenv("DEEPSEEK_API_KEY"), model="deepseek-chat")
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(DeepSeekWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-fireworks.py

@@ -0,0 +1,29 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.fireworks import FireworksLLMService
+
+load_dotenv(override=True)
+
+
+class FireworksWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = FireworksLLMService(
+            api_key=os.getenv("FIREWORKS_API_KEY"),
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(FireworksWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-gemini-multimodal.py

@@ -0,0 +1,38 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from multimodal_base_function_calling import MultimodalWeatherBot
+
+from pipecat.adapters.schemas.tools_schema import AdapterType
+from pipecat.services.gemini_multimodal_live import GeminiMultimodalLiveLLMService
+
+load_dotenv(override=True)
+
+
+class GeminiMultimodalWeatherBot(MultimodalWeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        search_tool = {"google_search": {}}
+        tools_def = MultimodalWeatherBot.tools()
+        tools_def.custom_tools = {AdapterType.GEMINI: [search_tool]}
+
+        llm = GeminiMultimodalLiveLLMService(
+            api_key=os.getenv("GOOGLE_API_KEY"),
+            voice_id="Puck",
+            transcribe_user_audio=True,
+            transcribe_model_audio=True,
+            tools=tools_def,
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(GeminiMultimodalWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-gemini.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.google import GoogleLLMService
+
+load_dotenv(override=True)
+
+
+class GeminiWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(GeminiWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-grok.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.grok import GrokLLMService
+
+load_dotenv(override=True)
+
+
+class GrokWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = GrokLLMService(api_key=os.getenv("GROK_API_KEY"))
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(GrokWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-groq.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.groq import GroqLLMService
+
+load_dotenv(override=True)
+
+
+class GroqWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = GroqLLMService(api_key=os.getenv("GROQ_API_KEY"), model="llama-3.3-70b-versatile")
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(GroqWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-nim.py

@@ -0,0 +1,29 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.nim import NimLLMService
+
+load_dotenv(override=True)
+
+
+class NimWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = NimLLMService(
+            api_key=os.getenv("NVIDIA_API_KEY"), model="meta/llama-3.3-70b-instruct"
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(NimWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-openai-realtime.py

@@ -0,0 +1,43 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from multimodal_base_function_calling import MultimodalWeatherBot
+
+from pipecat.services.openai_realtime_beta import (
+    InputAudioTranscription,
+    OpenAIRealtimeBetaLLMService,
+    SessionProperties,
+    TurnDetection,
+)
+
+load_dotenv(override=True)
+
+
+class OpenAiRealTimeWeatherBot(MultimodalWeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        session_properties = SessionProperties(
+            input_audio_transcription=InputAudioTranscription(),
+            # Set openai TurnDetection parameters. Not setting this at all will turn it
+            # on by default
+            turn_detection=TurnDetection(silence_duration_ms=1000),
+        )
+
+        llm = OpenAIRealtimeBetaLLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            session_properties=session_properties,
+            start_audio_paused=False,
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(OpenAiRealTimeWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-openai.py

@@ -0,0 +1,27 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.openai import OpenAILLMService
+
+load_dotenv(override=True)
+
+
+class OpenAiWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(OpenAiWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-openrouter.py

@@ -0,0 +1,29 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.openrouter import OpenRouterLLMService
+
+load_dotenv(override=True)
+
+
+class OpenRouterWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = OpenRouterLLMService(
+            api_key=os.getenv("OPENROUTER_API_KEY"), model="openai/gpt-4o-2024-11-20"
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(OpenRouterWeatherBot().run())

examples/unified-format-function-calling/standard-function-calling-together.py

@@ -0,0 +1,30 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from base_function_calling import WeatherBot
+from dotenv import load_dotenv
+
+from pipecat.services.together import TogetherLLMService
+
+load_dotenv(override=True)
+
+
+class TogetherWeatherBot(WeatherBot):
+    """Main class defining the LLM and passing it to the base handler."""
+
+    def __init__(self):
+        llm = TogetherLLMService(
+            api_key=os.getenv("TOGETHER_API_KEY"),
+            model="meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
+        )
+        super().__init__(llm)
+
+
+if __name__ == "__main__":
+    asyncio.run(TogetherWeatherBot().run())

scripts/fix-ruff.sh

@@ -0,0 +1,4 @@
+ruff format src
+ruff format examples
+ruff format tests
+ruff check --select I --fix

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)
 

src/pipecat/utils/test_frame_processor.py

@@ -8,6 +8,8 @@ class TestException(Exception):
 
 
 class TestFrameProcessor(FrameProcessor):
+    __test__ = False  # Prevents pytest from collecting this class as a test
+
     def __init__(self, test_frames):
         self.test_frames = test_frames
         self._list_counter = 0

tests/integration/test_integration_unified_function_calling.py

@@ -0,0 +1,96 @@
+import os
+from unittest.mock import AsyncMock
+
+import pytest
+from dotenv import load_dotenv
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.frames.frames import (
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMTextFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_services import LLMService
+from pipecat.services.anthropic import AnthropicLLMService
+from pipecat.services.google import GoogleLLMService
+from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService
+from pipecat.utils.test_frame_processor import TestFrameProcessor
+
+load_dotenv(override=True)
+
+
+def standard_tools() -> ToolsSchema:
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location"],
+    )
+    tools_def = ToolsSchema(standard_tools=[weather_function])
+    return tools_def
+
+
+async def _test_llm_function_calling(llm: LLMService):
+    # Create an AsyncMock for the function
+    mock_fetch_weather = AsyncMock()
+
+    llm.register_function(None, mock_fetch_weather)
+    t = TestFrameProcessor([LLMFullResponseStartFrame, LLMTextFrame, LLMFullResponseEndFrame])
+    llm.link(t)
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful assistant who can report the weather in any location in the universe. Respond concisely. Your response will be turned into speech so use only simple words and punctuation.",
+        },
+        {"role": "user", "content": " How is the weather today in San Francisco, California?"},
+    ]
+    context = OpenAILLMContext(messages, standard_tools())
+    # This is done by default inside the create_context_aggregator
+    context.set_llm_adapter(llm.get_llm_adapter())
+
+    frame = OpenAILLMContextFrame(context)
+
+    # This will fail if an exception is raised
+    await llm.process_frame(frame, FrameDirection.DOWNSTREAM)
+
+    # Assert that the mock function was called
+    mock_fetch_weather.assert_called_once()
+
+
+@pytest.mark.skipif(os.getenv("OPENAI_API_KEY") is None, reason="OPENAI_API_KEY is not set")
+@pytest.mark.asyncio
+async def test_unified_function_calling_openai():
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+    # This will fail if an exception is raised
+    await _test_llm_function_calling(llm)
+
+
+@pytest.mark.skipif(os.getenv("GOOGLE_API_KEY") is None, reason="GOOGLE_API_KEY is not set")
+@pytest.mark.asyncio
+async def test_unified_function_calling_gemini():
+    llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY"), model="gemini-2.0-flash-001")
+    # This will fail if an exception is raised
+    await _test_llm_function_calling(llm)
+
+
+@pytest.mark.skipif(os.getenv("ANTHROPIC_API_KEY") is None, reason="ANTHROPIC_API_KEY is not set")
+@pytest.mark.asyncio
+async def test_unified_function_calling_anthropic():
+    llm = AnthropicLLMService(
+        api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20240620"
+    )
+    # This will fail if an exception is raised
+    await _test_llm_function_calling(llm)

tests/test_function_calling_adapters.py

@@ -0,0 +1,176 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import unittest
+
+from openai.types.chat import ChatCompletionToolParam
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
+from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
+from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
+from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
+
+
+class TestFunctionAdapters(unittest.TestCase):
+    def setUp(self) -> None:
+        """Sets up a common tools schema for all tests."""
+        function_def = FunctionSchema(
+            name="get_weather",
+            description="Get the weather in a given location",
+            properties={
+                "location": {"type": "string", "description": "The city, e.g. San Francisco"},
+                "format": {
+                    "type": "string",
+                    "enum": ["celsius", "fahrenheit"],
+                    "description": "The temperature unit to use.",
+                },
+            },
+            required=["location", "format"],
+        )
+        self.tools_def = ToolsSchema(standard_tools=[function_def])
+
+    def test_openai_adapter(self):
+        """Test OpenAI adapter format transformation."""
+        expected = [
+            ChatCompletionToolParam(
+                type="function",
+                function={
+                    "name": "get_weather",
+                    "description": "Get the weather in a given location",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "location": {
+                                "type": "string",
+                                "description": "The city, e.g. San Francisco",
+                            },
+                            "format": {
+                                "type": "string",
+                                "enum": ["celsius", "fahrenheit"],
+                                "description": "The temperature unit to use.",
+                            },
+                        },
+                        "required": ["location", "format"],
+                    },
+                },
+            )
+        ]
+        assert OpenAILLMAdapter().to_provider_tools_format(self.tools_def) == expected
+
+    def test_anthropic_adapter(self):
+        """Test Anthropic adapter format transformation."""
+        expected = [
+            {
+                "name": "get_weather",
+                "description": "Get the weather in a given location",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {
+                        "location": {
+                            "type": "string",
+                            "description": "The city, e.g. San Francisco",
+                        },
+                        "format": {
+                            "type": "string",
+                            "enum": ["celsius", "fahrenheit"],
+                            "description": "The temperature unit to use.",
+                        },
+                    },
+                    "required": ["location", "format"],
+                },
+            }
+        ]
+        assert AnthropicLLMAdapter().to_provider_tools_format(self.tools_def) == expected
+
+    def test_gemini_adapter(self):
+        """Test Gemini adapter format transformation."""
+        expected = [
+            {
+                "function_declarations": [
+                    {
+                        "name": "get_weather",
+                        "description": "Get the weather in a given location",
+                        "parameters": {
+                            "type": "object",
+                            "properties": {
+                                "location": {
+                                    "type": "string",
+                                    "description": "The city, e.g. San Francisco",
+                                },
+                                "format": {
+                                    "type": "string",
+                                    "enum": ["celsius", "fahrenheit"],
+                                    "description": "The temperature unit to use.",
+                                },
+                            },
+                            "required": ["location", "format"],
+                        },
+                    }
+                ]
+            }
+        ]
+        assert GeminiLLMAdapter().to_provider_tools_format(self.tools_def) == expected
+
+    def test_openai_realtime_adapter(self):
+        """Test Anthropic adapter format transformation."""
+        expected = [
+            {
+                "type": "function",
+                "name": "get_weather",
+                "description": "Get the weather in a given location",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "location": {
+                            "type": "string",
+                            "description": "The city, e.g. San Francisco",
+                        },
+                        "format": {
+                            "type": "string",
+                            "enum": ["celsius", "fahrenheit"],
+                            "description": "The temperature unit to use.",
+                        },
+                    },
+                    "required": ["location", "format"],
+                },
+            }
+        ]
+        assert OpenAIRealtimeLLMAdapter().to_provider_tools_format(self.tools_def) == expected
+
+    def test_gemini_adapter_with_custom_tools(self):
+        """Test Gemini adapter format transformation."""
+        search_tool = {"google_search": {}}
+        expected = [
+            {
+                "function_declarations": [
+                    {
+                        "name": "get_weather",
+                        "description": "Get the weather in a given location",
+                        "parameters": {
+                            "type": "object",
+                            "properties": {
+                                "location": {
+                                    "type": "string",
+                                    "description": "The city, e.g. San Francisco",
+                                },
+                                "format": {
+                                    "type": "string",
+                                    "enum": ["celsius", "fahrenheit"],
+                                    "description": "The temperature unit to use.",
+                                },
+                            },
+                            "required": ["location", "format"],
+                        },
+                    }
+                ]
+            },
+            search_tool,
+        ]
+        tools_def = self.tools_def
+        tools_def.custom_tools = {AdapterType.GEMINI: [search_tool]}
+        assert GeminiLLMAdapter().to_provider_tools_format(tools_def) == expected