Function Tools

from zrt import Agent, function_tool

# External tool defined outside the class
@function_tool
def get_weather(location: str) -> str:
    """Get weather information for a specific location."""
    return f"Weather in {location}: Sunny, 72°F"

class WeatherAgent(Agent):
    def __init__(self, pipeline):
        super().__init__(
            agent_id="weather-assistant",
            instructions="You are a weather assistant.",
            pipeline=pipeline,
            tools=[get_weather],  # Register the external tool
        )

from zrt import Agent, function_tool

class FinanceAgent(Agent):
    def __init__(self, pipeline):
        super().__init__(
            agent_id="finance-assistant",
            instructions="You are a helpful financial assistant.",
            pipeline=pipeline,
        )
        self.portfolio = {"AAPL": 10, "GOOG": 5}

    @function_tool
    def get_portfolio_value(self) -> dict:
        """Get the current value of the user's stock portfolio."""
        # Access agent state via self
        return {"total_value": 5000, "holdings": self.portfolio}

from zrt import Pipeline, ContextWindow

pipeline = Pipeline(
    # ...stt, llm, tts, vad, turn_detector
    context_window=ContextWindow(
        max_tool_calls_per_turn=10,  # Allow up to 10 tool calls per turn
    ),
)

# add a tool
agent.update_tools(agent.tools + [get_horoscope])
# remove a tool
agent.update_tools([t for t in agent.tools if t is not get_horoscope])
# replace all tools
agent.update_tools([get_horoscope])

import asyncio
from zrt import Agent

class VoiceAgent(Agent):
    def __init__(self, pipeline):
        super().__init__(
            agent_id="assistant",
            instructions="You are a helpful voice assistant.",
            pipeline=pipeline,
        )

    async def on_enter(self):
        await self.session.say("Hi! How can I help?")

        async def update_tools_later():
            await asyncio.sleep(20)
            # Apply the change to the live session after a delay.
            self.update_tools(self.tools + [get_horoscope])

        asyncio.create_task(update_tools_later())