VideoToolkit

https://github.com/googleapis/python-genai https://ai.google.dev/gemini-api/docs/api-key

VideoToolkit

Bases: AsyncBaseToolkit

Source code in utu/tools/video_toolkit.py

class VideoToolkit(AsyncBaseToolkit):
    def __init__(self, config: ToolkitConfig = None) -> None:
        super().__init__(config)
        self.client = genai.Client(
            api_key=self.config.config.get("google_api_key"), http_options=HttpOptions(api_version="v1alpha")
        )
        self.model = self.config.config.get("google_model")

    @register_tool
    async def video_qa(self, video_url: str, question: str) -> str:
        r"""Asks a question about the video.

        Args:
            video_url (str): The path or URL to the video file.
            question (str): The question to ask about the video.
        """
        if not video_url.startswith("http"):
            video_part = Part.from_uri(file_uri=video_url)
        else:
            # e.g. Youtube URL
            video_part = Part.from_uri(
                file_uri=video_url,
                mime_type="video/mp4",
            )
        response = self.client.models.generate_content(
            model=self.model,
            contents=[
                question,
                video_part,
            ],
        )

        logger.debug(f"Video analysis response from gemini: {response.text}")
        return response.text

tools_map property

tools_map: dict[str, Callable]

Lazy loading of tools map. - collect tools registered by @register_tool

video_qa async

video_qa(video_url: str, question: str) -> str

Asks a question about the video.

Parameters:

Name	Type	Description	Default
video_url	str	The path or URL to the video file.	required
question	str	The question to ask about the video.	required

Source code in utu/tools/video_toolkit.py

@register_tool
async def video_qa(self, video_url: str, question: str) -> str:
    r"""Asks a question about the video.

    Args:
        video_url (str): The path or URL to the video file.
        question (str): The question to ask about the video.
    """
    if not video_url.startswith("http"):
        video_part = Part.from_uri(file_uri=video_url)
    else:
        # e.g. Youtube URL
        video_part = Part.from_uri(
            file_uri=video_url,
            mime_type="video/mp4",
        )
    response = self.client.models.generate_content(
        model=self.model,
        contents=[
            question,
            video_part,
        ],
    )

    logger.debug(f"Video analysis response from gemini: {response.text}")
    return response.text

get_tools_map_func

get_tools_map_func() -> dict[str, Callable]

Get tools map. It will filter tools by config.activated_tools if it is not None.

Source code in utu/tools/base.py

def get_tools_map_func(self) -> dict[str, Callable]:
    """Get tools map. It will filter tools by config.activated_tools if it is not None."""
    if self.config.activated_tools:
        assert all(tool_name in self.tools_map for tool_name in self.config.activated_tools), (
            f"Error config activated tools: {self.config.activated_tools}! available tools: {self.tools_map.keys()}"
        )
        tools_map = {tool_name: self.tools_map[tool_name] for tool_name in self.config.activated_tools}
    else:
        tools_map = self.tools_map
    return tools_map

get_tools_in_agents

get_tools_in_agents() -> list[FunctionTool]

Get tools in openai-agents format.

Source code in utu/tools/base.py

def get_tools_in_agents(self) -> list[FunctionTool]:
    """Get tools in openai-agents format."""
    tools_map = self.get_tools_map_func()
    tools = []
    for _, tool in tools_map.items():
        tools.append(
            function_tool(
                tool,
                strict_mode=False,  # turn off strict mode
            )
        )
    return tools

get_tools_in_openai

get_tools_in_openai() -> list[dict]

Get tools in OpenAI format.

Source code in utu/tools/base.py

def get_tools_in_openai(self) -> list[dict]:
    """Get tools in OpenAI format."""
    tools = self.get_tools_in_agents()
    return [ChatCompletionConverter.tool_to_openai(tool) for tool in tools]

get_tools_in_mcp

get_tools_in_mcp() -> list[Tool]

Get tools in MCP format.

Source code in utu/tools/base.py

def get_tools_in_mcp(self) -> list[types.Tool]:
    """Get tools in MCP format."""
    tools = self.get_tools_in_agents()
    return [MCPConverter.function_tool_to_mcp(tool) for tool in tools]

call_tool async

call_tool(name: str, arguments: dict) -> str

Call a tool by its name.

Source code in utu/tools/base.py

async def call_tool(self, name: str, arguments: dict) -> str:
    """Call a tool by its name."""
    tools_map = self.get_tools_map_func()
    if name not in tools_map:
        raise ValueError(f"Tool {name} not found")
    tool = tools_map[name]
    return await tool(**arguments)