import datetime
from livekit.agents import function_tool

@function_tool
async def get_current_datetime(output_format: str = None) -> str:
    """Get the current date and time in the given format.

    Args:
        output_format: formatting string for the date and time, defaults to '%Y-%m-%d %H:%M:%S'
    """
    return datetime.datetime.now().strftime(output_format or "%Y-%m-%d %H:%M:%S")

import asyncio
from typing import Any
from ddgs import DDGS
from livekit.agents import function_tool, ToolError

@function_tool
async def search_web(query: str, max_results: int = 1) -> dict[str, Any]:
    """
    Search the web using DuckDuckGo search engine for information about a given query.

    Args:
        query (str): The query to search for.
        max_results (int): The maximum number of results to return.

    Returns:
        dict[str, Any]: A dictionary containing the search results.
        The keys are the index of the result and the values are another dictionary with the following keys:
        - title: Title of the result.
        - url: URL of the result.
        - body: Body of the result.
    """
    try:
        g = DDGS()
        results = g.text(query, max_results=max_results)
    except Exception as e:
        return ToolError(f"Error searching the web: {e}")
    
    d = {str(i): res for i, res in enumerate(results)}
    for v in d.values():
        v["url"] = v.pop("href")
    return d

import asyncio
import datetime
from dataclasses import dataclass
from typing import Any
from ddgs import DDGS
from dotenv import load_dotenv

from livekit.agents import (
    Agent,
    AgentSession,
    JobContext,
    RunContext,
    ToolError,
    WorkerOptions,
    cli,
    function_tool,
)
from livekit.plugins import openai

load_dotenv()

@dataclass
class AppData:
    ddgs_client: DDGS

class MyAgent(Agent):
    def __init__(self):
        super().__init__(
            instructions="You are a helpful assistant with access to current time and web search capabilities."
        )

    @function_tool
    async def get_current_datetime(self, output_format: str = None) -> str:
        """Get the current date and time in the given format.

        Args:
            output_format: formatting string for the date and time, defaults to '%Y-%m-%d %H:%M:%S'
        """
        return datetime.datetime.now().strftime(output_format or "%Y-%m-%d %H:%M:%S")

    @function_tool
    async def search_web(self, ctx: RunContext[AppData], query: str) -> dict[str, Any]:
        """
        Performs a web search using the DuckDuckGo search engine.

        Args:
            query: The search term or question you want to look up online.
        """
        ddgs_client = ctx.userdata.ddgs_client
        
        # using asyncio.to_thread because the DDGS client is not asyncio compatible
        search = await asyncio.to_thread(ddgs_client.text, query)
        if len(search) == 0:
            raise ToolError("Tell the user that no results were found for the query.")

        return search

async def entrypoint(ctx: JobContext):
    app_data = AppData(ddgs_client=DDGS())
    
    agent = MyAgent()
    session = AgentSession(llm=openai.realtime.RealtimeModel(), userdata=app_data)

    await session.start(agent=agent, room=ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

@function_tool