[Python] How to access an uploaded video/article from google adk web using adk tool

I'm developing an application using ADK Web (specifically with InMemoryService for artifact and session management) and am encountering difficulties accessing video files uploaded by the user. My goal is to retrieve and process these uploaded video files within my ADK Web application. I've attempted several approaches, but none have been successful so far. What I've Tried:

tool_context.load_artifact(): My primary attempt was to use

`video_part = await tool_context.load_artifact(filename=video_filename)`


However, this method does not seem to return the expected video data. I'm unsure if load_artifact is the correct function for retrieving user-uploaded files, or if there's a specific way to handle video file types with it.

Accessing Data as Parameters: I've explored passing the video data directly as a function parameter, trying various types: Dict Type Part Type (assuming this might be relevant for file uploads) base64 string (to embed the video content directly) None of these methods allowed me to successfully access or reconstruct the video file. |

Research Undertaken: I've reviewed the official ADK Web documentation, particularly sections related to artifact management, session handling, and file uploads. Despite my efforts, I haven't found any clear examples or specific guidance on how to programmatically access files that users upload through the ADK Web interface when InMemoryService is in use.

code

import base64
import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent
from google.cloud import videointelligence
import os
import shutil
from google.adk.artifacts import InMemoryArtifactService
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService
from google.adk.tools import ToolContext
from google.adk.runners import Runner
from google.adk.tools import FunctionTool
from google.genai import types
from typing import Optional, Dict, Any
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

def save_video_to_current_dir(filename: str) -> str:
"""
Save the video from ADK web UI to current directory.

Args:
filename: The filename of the video from ADK web UI

Returns:
The path where the video was saved
"""
try:
# Get the current directory
current_dir = os.getcwd()

# Create a new filename with timestamp
timestamp = datetime.datetime.now(ZoneInfo('UTC')).strftime('%Y%m%d_%H%M%S')
new_filename = f"input_video_{timestamp}.mp4"

# Full path for the new file
output_path = os.path.join(current_dir, new_filename)

# Copy the file from ADK web UI location to current directory
shutil.copy2(filename, output_path)

logger.info(f"Successfully saved video to: {output_path}")
return output_path

except Exception as e:
logger.error(f"Error saving video: {str(e)}")
raise

async def process_video_input(video_data: Dict[str, Any], tool_context: ToolContext) -> Dict[str, Any]:
"""
Process video input from ADK web UI.

Args:
video_data: Dictionary containing video data from the UI with a 'video' key containing the video filename
tool_context: The context object provided by the ADK framework

Returns:
A dictionary containing:
- status: A string indicating the status of video processing
- video: The processed video part (if successful)
"""
try:
logger.info("Received video input from ADK web UI")
logger.info(f"Video data received: {video_data}")

# Get the video filename from the input data
if 'video' not in video_data:
raise ValueError("No video data found in input")

video_filename = video_data['video']
logger.info(f"Processing video with filename: {video_filename}")

# Load the video content using tool_context
video_part = await tool_context.load_artifact(filename=video_filename)
if not video_part:
raise ValueError(f"Could not load video content for {video_filename}")

logger.info(f"Successfully loaded video content")

# Save as artifact with a new name
timestamp = datetime.datetime.now(ZoneInfo('UTC')).strftime('%Y%m%d_%H%M%S')
artifact_name = f"input_video_{timestamp}.mp4"

version = await tool_context.save_artifact(
filename=artifact_name,
artifact=video_part
)

logger.info(f"Successfully saved video as artifact: {artifact_name} (version: {version})")

return {
"status": f"Video successfully processed and saved as {artifact_name}",
"video": video_part
}

except Exception as e:
logger.error(f"Error processing video: {str(e)}")
return {
"status": f"Error processing video: {str(e)}",
"video": None
}

def create_troll_video(filename: str, tool_context: ToolContext) -> Optional[str]:
"""
Creates a troll video from the input video.

Args:
filename: The filename of the input video artifact
tool_context: The context object provided by the ADK framework

Returns:
A string indicating the status of video creation
"""
try:
# Load the input video artifact using the tool context's load_artifact method
input_video = tool_context.load_artifact(filename=filename)

if not input_video:
return f"Could not find video artifact: {filename}"

# Process the video (implement your video processing logic here)
# For now, we'll just return a success message
return f"Successfully processed video {filename}"

except Exception as e:
logger.error(f"Error creating troll video: {str(e)}")
return f"Error creating troll video: {str(e)}"

# Initialize services
artifact_service = InMemoryArtifactService()
session_service = InMemorySessionService()

# Create the agent with video processing capabilities
root_agent = Agent(
name="troll_generator",
model="gemini-2.0-flash",
description=(
"This agent creates troll videos based on user queries. "
"It accepts video input from the ADK web UI and processes it to create entertaining content."
),
instruction=(
"You are a video creation agent that can create troll videos based on user queries. "
"You can accept video input from the ADK web UI and process it to create entertaining content. "
"Before creating a troll video, you need to process the video input first. "
"When a user uploads a video, it will be available as a video part in the input. "
"You should process this video part and save it as an artifact before proceeding with video creation. "
"After processing, you should return both the status message and the processed video back to the user."
),
tools=[
FunctionTool(process_video_input)
]
)

# Initialize the runner
runner = Runner(
agent=root_agent,
app_name="video_agent_app",
session_service=session_service,
artifact_service=artifact_service
)

Continue reading...