Skip to content

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 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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")

    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

    async def get_tools_map(self) -> dict[str, Callable]:
        return {
            "video_qa": self.video_qa,
        }

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 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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 async 

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 
58
59
60
61
62
63
64
65
66
67
68
69
async 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.tools_map is None:
        self.tools_map = await self.get_tools_map()
    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 async 

get_tools_in_agents() -> list[FunctionTool]

Get tools in openai-agents format.

Source code in utu/tools/base.py 
71
72
73
74
75
76
77
78
79
80
81
82
async def get_tools_in_agents(self) -> list[FunctionTool]:
    """Get tools in openai-agents format."""
    tools_map = await 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 async 

get_tools_in_openai() -> list[dict]

Get tools in OpenAI format.

Source code in utu/tools/base.py 
84
85
86
87
async def get_tools_in_openai(self) -> list[dict]:
    """Get tools in OpenAI format."""
    tools = await self.get_tools_in_agents()
    return [ChatCompletionConverter.tool_to_openai(tool) for tool in tools]

get_tools_in_mcp async 

get_tools_in_mcp() -> list[Tool]

Get tools in MCP format.

Source code in utu/tools/base.py 
89
90
91
92
async def get_tools_in_mcp(self) -> list[types.Tool]:
    """Get tools in MCP format."""
    tools = await 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 
 94
 95
 96
 97
 98
 99
100
async def call_tool(self, name: str, arguments: dict) -> str:
    """Call a tool by its name."""
    tools_map = await 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)