Releases: imvickykumar999/ADK-Web
Releases · imvickykumar999/ADK-Web
Release list
ADK Tools
Flask App
import os
import asyncio
import json
import sqlite3
import secrets # Import for generating secure session IDs
from dotenv import load_dotenv
# FIX: Added 'Response' to the import list
from flask import Flask, request, jsonify, g, redirect, url_for, make_response, Response
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai.types import Content, Part
# NOTE: The 'instance.agent' module is assumed to be available in the execution environment.
# Ensure 'instance/agent.py' exists and exports a 'root_agent' instance for this to work.
try:
# If using ADK, this is where your agent definition lives
from instance.agent import root_agent
except ImportError:
print("WARNING: 'instance.agent' could not be imported. Agent functionality will be disabled.")
root_agent = None
# Load environment variables from .env file
load_dotenv()
# --- ADK Initialization & Global State ---
APP_NAME = "agent_flask"
USER_ID = "web_user" # Keeping a fixed user ID for this web demo
# --- Database Configuration ---
DATABASE = 'history.db'
# Initialize Flask App EARLY to ensure it's available for decorators
app = Flask(__name__)
# --- Database Functions ---
def get_db():
"""Returns a database connection, creating one if not present in flask.g."""
if 'db' not in g:
g.db = sqlite3.connect(DATABASE)
g.db.row_factory = sqlite3.Row # Allows accessing columns by name
return g.db
@app.teardown_appcontext
def close_db(e=None):
"""Closes the database connection at the end of the request."""
db = g.pop('db', None)
if db is not None:
db.close()
def init_db():
"""Initializes the database and creates the messages table."""
with app.app_context():
db = get_db()
# Create a table to store chat messages
db.execute("""
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id TEXT NOT NULL,
role TEXT NOT NULL, -- 'user' or 'agent'
text TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
)
""")
db.commit()
def save_message(session_id: str, role: str, text: str):
"""Saves a single message to the database."""
try:
db = get_db()
db.execute(
"INSERT INTO messages (session_id, role, text) VALUES (?, ?, ?)",
(session_id, role, text)
)
db.commit()
except Exception as e:
app.logger.error(f"Database Save Error: {e}")
def load_history(session_id: str) -> list[dict]:
"""Loads all messages for a given session ID."""
try:
db = get_db()
rows = db.execute(
"SELECT role, text FROM messages WHERE session_id = ? ORDER BY timestamp ASC",
(session_id,)
).fetchall()
return [{"role": row['role'], "text": row['text']} for row in rows]
except Exception as e:
app.logger.error(f"Database Load Error: {e}")
return []
def get_all_session_ids() -> list[str]:
"""Loads all unique session IDs from the database."""
try:
db = get_db()
rows = db.execute(
# Order by timestamp of the last message in that session for better display logic
"""
SELECT T1.session_id FROM messages T1
INNER JOIN (
SELECT session_id, MAX(timestamp) AS max_timestamp
FROM messages
GROUP BY session_id
) T2 ON T1.session_id = T2.session_id AND T1.timestamp = T2.max_timestamp
GROUP BY T1.session_id
ORDER BY T2.max_timestamp DESC
"""
).fetchall()
return [row['session_id'] for row in rows]
except Exception as e:
app.logger.error(f"Database Session Load Error: {e}")
return []
# Initialize in-memory session service
session_service = InMemorySessionService()
# Create the runner with the agent only if root_agent was successfully imported
runner = None
adk_sessions = {} # Dictionary to track which ADK sessions have been created
if root_agent:
runner = Runner(
agent=root_agent,
app_name=APP_NAME,
session_service=session_service,
)
async def initialize_adk_session(session_id: str):
"""Creates the ADK session if it doesn't exist."""
if session_id not in adk_sessions:
app.logger.info(f"Creating ADK session for {USER_ID}/{session_id}")
await session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=session_id
)
adk_sessions[session_id] = True
# --- Helper to get/create session ID from request ---
def get_or_create_session_id():
"""Gets the session ID from the request or generates a new one."""
session_id = request.args.get('session_id')
if not session_id:
# Generate a new, short, URL-safe session ID (e.g., 'a3b7c4d8')
session_id = secrets.token_hex(4)
# Redirect to the new URL with the session_id query parameter
return redirect(url_for('index', session_id=session_id))
return session_id
# --- API Endpoints ---
@app.route('/history', methods=['GET'])
def get_history_api():
"""Returns the chat history and all sessions for the current session ID."""
current_session_id = request.args.get('session_id')
if not current_session_id:
return jsonify({"history": [], "sessions": []}), 200
history = load_history(current_session_id)
sessions = get_all_session_ids()
return jsonify({
"history": history,
"current_session_id": current_session_id,
"sessions": sessions
})
@app.route('/chat', methods=['POST'])
def chat():
"""Handles incoming user messages, runs the ADK agent, and returns the response."""
current_session_id = request.args.get('session_id')
if not current_session_id:
return jsonify({"response": "Error: Session ID is missing."}), 400
if not runner:
return jsonify({"response": "Error: Agent runner is not initialized. Check server logs."}), 500
# Ensure the ADK session is initialized for this ID
if root_agent and current_session_id not in adk_sessions:
try:
# Synchronously call the async session creator
asyncio.run(initialize_adk_session(current_session_id))
except Exception as e:
app.logger.error(f"ADK Session Initialization Error: {e}")
return jsonify({"response": f"ADK Session Init Error: {str(e)}"}), 500
data = request.get_json()
user_input = data.get('message', '').strip()
if not user_input:
return jsonify({"response": "Please provide a message."}), 400
# 1. Save user message to history
save_message(current_session_id, "user", user_input)
# Prepare the message for the runner
message = Content(role="user", parts=[Part(text=user_input)])
response_text = "Sorry, I encountered an internal error."
async def get_agent_response(msg, session_id):
"""Asynchronously runs the agent and extracts the final text response."""
response = ""
try:
async for event in runner.run_async(
user_id=USER_ID,
session_id=session_id,
new_message=msg
):
if hasattr(event, "is_final_response") and event.is_final_response():
if hasattr(event, "content") and event.content.parts:
# Extract text from the first part of the content
response = event.content.parts[0].text
break
except Exception as e:
# Handle potential ADK/Runner exceptions
return f"An agent error occurred: {str(e)}"
return response
try:
final_response = asyncio.run(get_agent_response(message, current_session_id))
if final_response.startswith("An agent error occurred"):
response_text = final_response
status_code = 500
else:
response_text = final_response
status_code = 200
# 2. Save agent message to history on success
save_message(current_session_id, "agent", response_text)
except Exception as e:
response_text = f"Flask runtime error: {str(e)}"
status_code = 500
return jsonify({"response": response_text}), status_code
@app.route('/')
def index():
"""Handles dynamic session creation and serves the main chat application page."""
# 1. Check for/create session_id and handle redirect if needed
session_id_result = get_or_create_session_id()
# FIX: Check against the Flask Response class type, as flask.redirect() returns a Response object.
if isinstance(session_id_result, Response):
return session_id_result
current_session_id = session_id_result
# 2. Pass the current session ID to the HTML generator
html_content = get_html_content(current_session_id)
response = make_response(html_content)
return response
# --- Frontend HTML/JS/CSS (Inlined for single-file deployment) ---
def get_html_content(current_session_id):
"""Generates the single HTML page with inline CSS and JavaScript for the chat UI."""
html_template = f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" ...Telegram ChatBot
session_service_uri
Steps to run:
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
adk web --session_service_uri sqlite:///sessions.db --port 5000
Go to http://localhost:5000
Ctrl + C to exit.
Database Session Service
>>> tree -la
.
├── .env
├── main.py
├── memory_agent
│ ├── agent.py
│ ├── __init__.py
│ └── __pycache__
│ ├── agent.cpython-312.pyc
│ └── __init__.cpython-312.pyc
├── my_agent_data.db
├── __pycache__
│ └── utils.cpython-312.pyc
└── utils.py
4 directories, 9 files>>> python3 main.py
Continuing existing session: 79177cc0-2736-417c-a4c3-bb0c9e6e4ded
Welcome to Memory Agent Chat!
Your reminders will be remembered across conversations.
Type 'exit' or 'quit' to end the conversation.
You: hello
--- Running Query: hello ---
---------- State BEFORE processing ----------
👤 User: Brandon Hancock
📝 Reminders: None
---------------------------------------------
Warning: there are non-text parts in the response: ['function_call'], returning concatenated text result from text parts. Check the full candidates.content.parts accessor to get the full model response.
Event ID: a1ee133a-6534-40b4-ba0e-f5f0f234a9dd, Author: memory_agent
--- Tool: add_reminder called for 'take out trash tomorrow at 9am' ---
Event ID: 8d5f7b3d-2711-495e-acd0-e02c65dd04b5, Author: memory_agent
Event ID: 6bd9fab4-9f80-4743-991a-2f5a6cb1a766, Author: memory_agent
Text: 'Hello Brandon, I've added 'take out trash tomorrow at 9am' to your reminders. I can remember your information across conversations!'
╔══ AGENT RESPONSE ═════════════════════════════════════════
Hello Brandon, I've added 'take out trash tomorrow at 9am' to your reminders. I can remember your information across conversations!
╚═════════════════════════════════════════════════════════════
---------- State AFTER processing ----------
👤 User: Brandon Hancock
📝 Reminders:
1. take out trash tomorrow at 9am
--------------------------------------------
You: remind me today at 8pm to order cake from zomato, got it?
--- Running Query: remind me today at 8pm to order cake from zomato, got it? ---
---------- State BEFORE processing ----------
👤 User: Brandon Hancock
📝 Reminders:
1. take out trash tomorrow at 9am
---------------------------------------------
Warning: there are non-text parts in the response: ['function_call'], returning concatenated text result from text parts. Check the full candidates.content.parts accessor to get the full model response.
Event ID: 6db441bd-28a5-4446-bfc3-cd892158beaa, Author: memory_agent
--- Tool: add_reminder called for 'order cake from zomato' ---
Event ID: d50d10a6-ab78-43b4-be8f-d33bc2c3d7c1, Author: memory_agent
Event ID: b597c0ef-db34-4319-85e5-24746ea1b4d8, Author: memory_agent
Text: 'Got it, Brandon! I've added 'order cake from zomato' to your reminders.'
╔══ AGENT RESPONSE ═════════════════════════════════════════
Got it, Brandon! I've added 'order cake from zomato' to your reminders.
╚═════════════════════════════════════════════════════════════
---------- State AFTER processing ----------
👤 User: Brandon Hancock
📝 Reminders:
1. take out trash tomorrow at 9am
2. order cake from zomato
--------------------------------------------
You: Web Search
Multi Agents
Navigate through multiple agents in dropdown, and select different sessions each with seperate memory history.
Do You Need Separate .env Files for Each Agent?
No, you don't strictly need three separate .env files (one in each agent folder like greeting_agent/, time_agent/, and tool_agent/). While the official Google ADK quickstart places a .env in each agent subfolder for convenience (e.g., to isolate configs if agents use different models or API keys), the framework relies on the python-dotenv library to load environment variables from .env files in the current working directory (CWD) or parent directories. Since you run commands like adk web or adk run <agent_name> from the parent directory (the root of your project, where all agent subfolders live), a single .env at the root level is automatically loaded for all agents.
(.venv) $ tree -la
.
├── .env
├── greeting_agent
│ ├── __init__.py
│ ├── __pycache__
│ │ ├── __init__.cpython-310.pyc
│ │ ├── __init__.cpython-312.pyc
│ │ ├── agent.cpython-310.pyc
│ │ └── agent.cpython-312.pyc
│ └── agent.py
├── multi-agents.zip
├── time_agent
│ ├── __init__.py
│ ├── __pycache__
│ │ ├── __init__.cpython-310.pyc
│ │ ├── __init__.cpython-312.pyc
│ │ ├── agent.cpython-310.pyc
│ │ └── agent.cpython-312.pyc
│ └── agent.py
└── tool_agent
├── __init__.py
├── __pycache__
│ ├── __init__.cpython-310.pyc
│ ├── __init__.cpython-312.pyc
│ ├── agent.cpython-310.pyc
│ └── agent.cpython-312.pyc
└── agent.py
7 directories, 20 files