Video blob service
video_blob_service ¶
Protocol definition for video blob loading services.
This module defines the VideoBlobServiceProtocol that enables loading video files and converting them to multimodal content parts for agent execution. The protocol provides methods for validating video files and preparing them as Part objects for the ADK Content API.
| ATTRIBUTE | DESCRIPTION |
|---|---|
VideoBlobServiceProtocol | Protocol for video blob loading. TYPE: |
Examples:
Implement a video blob service:
from gepa_adk.ports import VideoBlobServiceProtocol
from gepa_adk.domain.models import VideoFileInfo
class MockVideoBlobService:
async def prepare_video_parts(
self,
video_paths: list[str],
) -> list[Any]:
# Return mock Part objects
return [{"mock": path} for path in video_paths]
def validate_video_file(
self,
video_path: str,
) -> VideoFileInfo:
return VideoFileInfo(
path=video_path,
size_bytes=1024,
mime_type="video/mp4",
)
Verify protocol compliance:
from gepa_adk.ports import VideoBlobServiceProtocol
service = MockVideoBlobService()
assert isinstance(service, VideoBlobServiceProtocol)
Note
The protocol uses list[Any] as the return type for prepare_video_parts() to avoid importing ADK types in the ports layer. Implementations in the adapters layer return google.genai.types.Part objects.
VideoBlobServiceProtocol ¶
Bases: Protocol
flowchart TD
gepa_adk.ports.video_blob_service.VideoBlobServiceProtocol[VideoBlobServiceProtocol]
click gepa_adk.ports.video_blob_service.VideoBlobServiceProtocol href "" "gepa_adk.ports.video_blob_service.VideoBlobServiceProtocol"
Protocol for loading video files as multimodal content parts.
Implementations provide video file I/O and validation logic, converting video files to Part objects for multimodal agent input.
The protocol defines an async method for loading multiple videos and a sync method for validating individual files.
Examples:
Implement a video blob service:
class SimpleVideoBlobService:
async def prepare_video_parts(
self,
video_paths: list[str],
) -> list[Any]:
# Load videos and return Part objects
parts = []
for path in video_paths:
self.validate_video_file(path)
# ... load and create Part
return parts
def validate_video_file(
self,
video_path: str,
) -> VideoFileInfo:
# Validate and return metadata
...
Verify protocol compliance:
from gepa_adk.ports import VideoBlobServiceProtocol
service = SimpleVideoBlobService()
assert isinstance(service, VideoBlobServiceProtocol)
Note
All implementations must provide both prepare_video_parts() and validate_video_file() methods. Video size is limited to 2GB per the Gemini API constraint. Only video/* MIME types are accepted.
Source code in src/gepa_adk/ports/video_blob_service.py
61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 | |
prepare_video_parts async ¶
Load video files and create Part objects for multimodal content.
Reads video files from disk and converts them to Part objects suitable for inclusion in ADK Content. Files are validated before loading.
| PARAMETER | DESCRIPTION |
|---|---|
video_paths | List of absolute paths to video files. Must be non-empty. All paths must point to existing video files under 2GB with video/* MIME types. TYPE: |
| RETURNS | DESCRIPTION |
|---|---|
list[Any] | List of Part objects, one per input path. Order is preserved |
list[Any] | such that output[i] corresponds to video_paths[i]. Part objects |
list[Any] | contain inline video data with appropriate MIME type. |
| RAISES | DESCRIPTION |
|---|---|
ValueError | If video_paths is empty. |
VideoValidationError | If any file is not found, exceeds 2GB, or has a non-video MIME type. |
PermissionError | If any file cannot be read due to permissions. |
OSError | If any file cannot be read due to I/O errors. |
Examples:
Load a single video:
Load multiple videos preserving order:
paths = ["/data/intro.mp4", "/data/main.mp4", "/data/outro.mp4"]
parts = await service.prepare_video_parts(paths)
assert len(parts) == 3
# parts[0] is intro, parts[1] is main, parts[2] is outro
Note
Operations include async file I/O for reading video bytes. Video files are validated before loading to provide early failure with clear error messages.
Source code in src/gepa_adk/ports/video_blob_service.py
validate_video_file ¶
Validate a video file and return its metadata.
Checks that the file exists, is within size limits, and has a valid video MIME type. Returns metadata on success.
| PARAMETER | DESCRIPTION |
|---|---|
video_path | Absolute path to the video file to validate. TYPE: |
| RETURNS | DESCRIPTION |
|---|---|
'VideoFileInfo' | VideoFileInfo containing validated metadata: |
'VideoFileInfo' |
|
'VideoFileInfo' |
|
'VideoFileInfo' |
|
| RAISES | DESCRIPTION |
|---|---|
VideoValidationError | If the file does not exist, exceeds 2GB, or has a non-video MIME type. The exception includes the video_path and constraint fields for debugging. |
Examples:
Validate a video file:
info = service.validate_video_file("/data/lecture.mp4")
print(f"Size: {info.size_bytes} bytes")
print(f"Type: {info.mime_type}")
Handle validation errors:
from gepa_adk.domain.exceptions import VideoValidationError
try:
info = service.validate_video_file("/missing.mp4")
except VideoValidationError as e:
print(f"Invalid: {e.video_path} - {e.constraint}")
Note
Operates synchronously for fast pre-validation without async context. File content is not read, only metadata is checked.