Best-practise for route to return last event #3332
Description
I have ADK (1.16.0) running on Cloud Run and using DatabaseSessionService as session service.
I want the ADK web ui, as well as an additional route to return only the text of the final event to a separate front-end application.
I've found a way myself, but I would like to know from the community what would be the best practice. Perhaps such a route could become of AdkWebServer one day, simplifying the life of many developers since I'm sure this is a very common use case :)
This is how I set up the main.py
# --- Set up ADK ---
# Get the directory where the agents are located
AGENT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "agents")
# Set up default eval managers
eval_sets_manager = LocalEvalSetsManager(agents_dir=AGENT_DIR)
eval_set_results_manager = LocalEvalSetResultsManager(agents_dir=AGENT_DIR)
# Initialize Agent Loader
agent_loader = AgentLoader(AGENT_DIR)
# Create ADK web server
adk_web_server = AdkWebServer(
agent_loader=agent_loader,
session_service=session_service,
memory_service=memory_service,
artifact_service=artifact_service,
credential_service=credential_service,
eval_sets_manager=eval_sets_manager,
eval_set_results_manager=eval_set_results_manager,
agents_dir=AGENT_DIR,
extra_plugins=None,
)
extra_fast_api_args = {}
extra_fast_api_args.update(
register_processors=register_processors_wrapper,
)
# --- ADK dev UI ---
cli_package_dir = Path(adk_cli_package.__file__).parent.resolve()
angular_dist_path = cli_package_dir / "browser"
extra_fast_api_args.update(
web_assets_dir=angular_dist_path,
)
# --- Create FastAPI app with ADK default routes for dev UI ---
app = adk_web_server.get_fast_api_app(
lifespan=lifespan,
allow_origins=app_settings.allowed_origins,
**extra_fast_api_args,
)
# --- Add custom routes ---
app.state.adk_web_server = adk_web_server
app.state.session_service = session_service
app.include_router(new_message_router)
And this is the route:
def get_adk_server(request: Request) -> AdkWebServer:
"""Dependency function to get the AdkWebServer from the app state."""
if not hasattr(request.app.state, "adk_web_server"):
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="ADK web server is not initialized.",
)
return request.app.state.adk_web_server
@router.post(path="/messages", response_model=Message)
async def generate_new_message_route(
message: NewMessage,
adk_web_server: AdkWebServer = Depends(get_adk_server),
) -> str:
"""Generate agent response for user request.
Return only the final agent response.
"""
# --- Structure message for ADK ---
parts_list = [types.Part(text=message.text)]
# Add optional attachment as a Part object
if message.user_file_gcs_uri and message.user_file_mime_type:
file_data = types.FileData(
file_uri=message.user_file_gcs_uri, mime_type=message.user_file_mime_type
)
parts_list.append(types.Part(file_data=file_data))
content_message = types.Content(role="user", parts=parts_list)
logger.debug(f"[new message] Sending Content to ADK: {content_message}")
# --- Call ADK ---
runner = await adk_web_server.get_runner_async(
app_name=app_settings.APP_NAME,
)
# Invoke response generation
async with Aclosing(
runner.run_async(
user_id=message.user_id,
session_id=message.session_id,
new_message=content_message,
)
) as agen:
events = [event async for event in agen]
logger.debug(f"[new message] Generated {len(events)} events in agent run")
logger.debug(f"[new message] Events generated: {events}")
# --- Process response ---
# Extract final event response
final_event = next(
(
event
for event in reversed(events)
if event.is_final_response
and event.content
and event.content.parts
and event.content.parts[0].text # Check if text is present
),
None, # Default to None if no final event is found
)
final_response_text = final_event.content.parts[0].text if final_event else None
if final_event and final_response_text:
logger.debug(f"[new message] Final response: {final_response_text}")
return final_response_text