Telegram bot

.env

@@ -0,0 +1,3 @@
+GOOGLE_API_KEY=AIzaSyCPF-QQ04wj6GdY9zIWqW46Z4HlXpccpOg
+GOOGLE_GENAI_USE_VERTEXAI=False
+TELEGRAM_BOT_TOKEN=8171811430:AAFlatEQrPmD7K4-2XHDlHDPVz1-Q4gCbms

.gitignore

@@ -0,0 +1,4 @@
+__pycache__/
+venv/
+.env
+tasks.db

README.md

@@ -1,13 +1,73 @@
----
-title: Example Example
-emoji: 🌖
-colorFrom: red
-colorTo: red
-sdk: gradio
-sdk_version: 5.49.1
-app_file: app.py
-pinned: false
-short_description: Exapmke app
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# PROJECT TITLE
+ **Task Manager Telegram bot:** An AI-powered Telegram bot that helps users add, edit, list, and delete tasks seamlessly using natural language or commands.
+
+---
+
+### PROJECT DESCRIPTION:
+Task Manager Telegram Bot is an AI-powered Telegram bot designed to help users manage their tasks efficiently. It allows users to:
+
+- Add new tasks using commands or free text.
+- Edit existing tasks with ease.
+- List all current tasks in an organized format.
+- Delete tasks individually with confirmation.
+- Keep track of task history and previous conversations.
+- Interact naturally using both commands and conversational language.
+
+---
+
+### TECHNOLOGIES USED:
+
+- **PYTHON:** The core programming language used to build the Telegram bot and implement all task management functionalities.
+
+- **PYTHON-TELEGRAM-BOT:** A Python library used to interact with the Telegram Bot API for sending and receiving messages.
+
+- **SQLALCHEMY & SQLITE:** Used for storing user tasks and chat history in a lightweight relational database.
+
+- **GOOGLE GEMINI (via gen-ai):** Provides AI-powered natural language understanding for interpreting free-text task commands and generating smart responses.
+
+- **GOOGLE ADK AGENT:** Used to create an intelligent agent that maps user messages to task management actions, enabling context-aware task handling and conversational AI capabilities.
+
+---
+### HOW TO GET TELEGRAM TOKEN:
+- Open Telegram and search for BotFather.
+- Start a chat with BotFather using the /start command.
+- Create a new bot using the /newbot command.
+- Follow the prompts to provide a name and username for your bot.
+- After successful creation, BotFather will give you a bot token.
+
+---
+
+### HOW TO INSTALL AND RUN PROJECT:
+- Ensure you have Python installed on your system.
+- Clone this repository to your local machine.
+  ```bash 
+    git clone <repository>
+- Open the project folder in your preferred code editor (e.g., VS Code).
+-Create a .env file in the root directory to store your Gemini API key:
+    ```bash
+        API_KEY=your_gemini_api_key
+        TELEGRAM_BOT_KEY=your_telegram_token
+- Navigate to the project directory in your terminal.
+- Create virtual environment and activate it:
+    ```bash
+        python -m venv venv
+        venv\Scripts\activate            
+- Install the dependancies:
+    ```bash
+        pip install -r requirements.txt
+- Run the application:
+    ```bash
+        python echobot.py 
+-----             
+### HOW TO USE THE PROJECT:
+
+- Open Telegram and search for your bot by its username.
+
+- Start a chat with the bot by clicking Start.
+
+- You can interact with the bot in two ways:
+    - Commands: Use predefined commands like /newtask, /listtask, /edittask, /deletetask to manage your tasks.
+    - Free text: Simply type messages like “remind me to buy milk” or “delete my meeting task”, and the bot will understand and perform the action.
+- The bot will respond with confirmations, task lists, or error messages as needed.        
+---
+

agent.py

@@ -0,0 +1,33 @@
+from google.adk.agents import Agent
+from tools import add_task, list_tasks, edit_task, delete_task
+
+task_manager_agent = Agent(
+    name="task_manager_agent",
+    model="gemini-2.5-flash",
+    instruction = """
+# TaskManagerAgent: Smart Telegram Task Assistant
+
+## Persona
+You are a task manager assistant. Users interact with you through **commands** or **free text**.  
+Your job is to always map user requests into the correct task management action.
+Take the telegram user_id from runner and if the user enters any tasks or performs any operations update everything to that user_id.
+
+## Core Rules
+1. If user sends a command (`/newtask`, `/listtask`, `/edittask`, `/deletetask`), use the tool directly.
+2. If user sends free text, interpret and call the matching tool:
+   - "remind me to buy milk" → call /newtask with task="buy milk".
+   - "show my tasks" → call /listtask.
+   - "delete my meeting task" → call /deletetask with task="meeting".
+   - "change drink juice to drink green tea" → call /edittask with old="drink juice", new="drink green tea".
+3. Always reply with a clear confirmation message:
+   - ✅ Task added: <task>
+   - 📋 Your tasks: <list>
+   - 🗑️ Task deleted: <task>
+   - ✏️ Task updated: <old → new>
+4. If request is unclear, ask for clarification instead of failing.
+5. Never reply with “Sorry, I didn't understand”.
+"""
+,
+
+    tools=[add_task, list_tasks, edit_task, delete_task]
+)

db_utils.py

@@ -0,0 +1,78 @@
+from sqlalchemy import create_engine, Column, Integer, String, DateTime
+from sqlalchemy.orm import sessionmaker, declarative_base
+from datetime import datetime
+
+Base = declarative_base()
+
+class ChatMessage(Base):
+    __tablename__ = "chat_messages"
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(String, index=True)
+    role = Column(String)  # "user" or "bot"
+    message = Column(String)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+
+class Task(Base):
+    __tablename__ = "tasks"
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(String, index=True)
+    name = Column(String)
+    timestamp = Column(DateTime, default=datetime.utcnow)
+
+engine = create_engine("sqlite:///tasks.db", echo=False)
+SessionLocal = sessionmaker(bind=engine)
+
+def init_db():
+    Base.metadata.create_all(bind=engine)
+
+def save_message(user_id, role, message):
+    session = SessionLocal()
+    msg = ChatMessage(user_id=user_id, role=role, message=message)
+    session.add(msg)
+    session.commit()
+    session.close()
+
+def load_chat_history(user_id, limit=5):
+    session = SessionLocal()
+    messages = session.query(ChatMessage).filter_by(user_id=user_id).order_by(ChatMessage.timestamp.desc()).limit(limit).all()
+    session.close()
+    return reversed(messages)
+
+# --- Task DB functions ---
+def add_task_db(user_id, task_name):
+    session = SessionLocal()
+    task = Task(user_id=user_id, name=task_name)
+    session.add(task)
+    session.commit()
+    session.close()
+    return f'Task "{task_name}" added successfully ✅'
+
+def list_tasks_db(user_id):
+    session = SessionLocal()
+    tasks = session.query(Task).filter_by(user_id=user_id).order_by(Task.timestamp.asc()).all()
+    session.close()
+    if not tasks:
+        return "No tasks found."
+    return "Your Tasks:\n" + "\n".join([f"{i+1}. {t.name}" for i, t in enumerate(tasks)])
+
+def edit_task_db(user_id, old_name, new_name):
+    session = SessionLocal()
+    task = session.query(Task).filter_by(user_id=user_id, name=old_name).first()
+    if task:
+        task.name = new_name
+        session.commit()
+        session.close()
+        return f'Task "{old_name}" updated to "{new_name}" ✅'
+    session.close()
+    return f'Task "{old_name}" not found ❌'
+
+def delete_task_db(user_id, task_name):
+    session = SessionLocal()
+    task = session.query(Task).filter_by(user_id=user_id, name=task_name).first()
+    if task:
+        session.delete(task)
+        session.commit()
+        session.close()
+        return f'Task "{task_name}" deleted successfully 🗑️'
+    session.close()
+    return f'Task "{task_name}" not found ❌'

echobot.py

@@ -0,0 +1,191 @@
+
+import logging
+import asyncio
+from dotenv import load_dotenv
+import os
+from telegram import Update, ForceReply
+from telegram.ext import (
+    Application, CommandHandler, MessageHandler, ContextTypes, filters, ConversationHandler
+)
+from tools import add_task, list_tasks, edit_task, delete_task
+from runner import create_session_async, run_agent
+from db_utils import save_message, load_chat_history, init_db
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
+)
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+session = None
+
+NEW_TASK, EDIT_TASK_NAME, EDIT_TASK_NEW_NAME, DELETE_TASK_NAME, DELETE_TASK_CONFIRM = range(5)
+
+async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    menu = (
+        "Welcome to Task Manager Bot!\n"
+        "Choose a method:\n"
+        "/newtask - Add a new task\n"
+        "/listtask - View all tasks\n"
+        "/edittask - Edit an existing task\n"
+        "/deletetask - Delete a task\n"
+        "/history - Show last 5 chats"
+    )
+    await update.message.reply_text(menu, reply_markup=ForceReply(selective=True))
+
+
+async def newtask_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    await update.message.reply_text("Enter task name:")
+    return NEW_TASK
+
+async def newtask_save(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    user_id = str(update.effective_user.id)
+    task_name = update.message.text.strip()
+    if task_name:
+        result = add_task(user_id, task_name)
+        await update.message.reply_text(result)
+    else:
+        await update.message.reply_text("Task name cannot be empty. Please try again.")
+        return NEW_TASK
+    return ConversationHandler.END
+
+async def listtask(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    user_id = str(update.effective_user.id)
+    result = list_tasks(user_id)
+    await update.message.reply_text(result)
+
+async def edittask_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    await update.message.reply_text("Enter the task name you want to edit:")
+    return EDIT_TASK_NAME
+
+async def edittask_get_new_name(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    user_id = str(update.effective_user.id)
+    old_name = update.message.text.strip()
+    context.user_data['edit_task_old_name'] = old_name
+
+    tasks_list = list_tasks(user_id)
+    if old_name not in tasks_list:
+        await update.message.reply_text("Task not found. Please enter a valid task name:")
+        return EDIT_TASK_NAME
+    await update.message.reply_text("Enter the new task name:")
+    return EDIT_TASK_NEW_NAME
+
+async def edittask_save(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    user_id = str(update.effective_user.id)
+    new_name = update.message.text.strip()
+    old_name = context.user_data.get('edit_task_old_name')
+    result = edit_task( user_id, old_name, new_name)
+    await update.message.reply_text(result)
+    return ConversationHandler.END
+
+async def deletetask_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    await update.message.reply_text("Enter the task name you want to delete:")
+    return DELETE_TASK_NAME
+
+async def deletetask_confirm(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    user_id = str(update.effective_user.id)
+    task_name = update.message.text.strip()
+    context.user_data['delete_task_name'] = task_name
+    tasks_list = list_tasks(user_id)
+    if task_name not in tasks_list:
+        await update.message.reply_text("Task not found. Please enter a valid task name:")
+        return DELETE_TASK_NAME
+    await update.message.reply_text(f'Are you sure you want to delete "{task_name}"? If yes, type "yes" else type /cancel')
+    return DELETE_TASK_CONFIRM
+
+async def deletetask_finish(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    user_id = str(update.effective_user.id)
+    response = update.message.text.strip().lower()
+    task_name = context.user_data.get('delete_task_name')
+    if response == "yes":
+        result = delete_task(user_id, task_name)
+        await update.message.reply_text(result)
+    else:
+        await update.message.reply_text("Delete cancelled ❌")
+    return ConversationHandler.END
+
+
+async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
+    await update.message.reply_text("Cancelled ❌")
+    return ConversationHandler.END
+
+async def history(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    user_id = str(update.effective_user.id)
+    messages = load_chat_history(user_id, limit=5)
+    if not messages:
+        await update.message.reply_text("No history yet.")
+        return
+    history_text = "\n".join([f"[{m.role}] {m.message}" for m in messages])
+    await update.message.reply_text("🕑 Your recent chats:\n\n" + history_text)
+
+async def handle_free_text(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
+    user_id = int(update.effective_user.id)
+    user_message = update.message.text.strip()
+    save_message(user_id, "user", user_message)
+    await update.message.reply_text(f"🤖 Thinking... {user_id}")
+    try:
+        response = await run_agent(user_id, session.id, user_message)
+        if not response:
+            response_text = "⚠️ I didn't get any response."
+        elif isinstance(response, dict):
+            response_text = response.get("text") or response.get("message") or str(response)
+        else:
+            response_text = str(response).strip()
+        # if not response_text:
+        #     response_text = "⚠️ Sorry, I got an empty reply."
+
+        save_message(user_id, "bot", response_text)
+        await update.message.reply_text(response_text)
+
+    except Exception as e:
+        await update.message.reply_text(f"❌ Error: {str(e)}")
+
+
+
+def main():
+    global session
+    init_db()
+    loop = asyncio.get_event_loop()
+    session = loop.run_until_complete(create_session_async())
+    telegram_token = os.getenv("TELEGRAM_BOT_TOKEN") 
+    application = Application.builder().token(telegram_token).build()
+    application.add_handler(CommandHandler("start", start))
+    application.add_handler(CommandHandler("listtask", listtask))
+    application.add_handler(CommandHandler("history", history))
+
+    newtask_conv = ConversationHandler(
+        entry_points=[CommandHandler("newtask", newtask_start)],
+        states={NEW_TASK: [MessageHandler(filters.TEXT & ~filters.COMMAND, newtask_save)]},
+        fallbacks=[]
+    )
+    application.add_handler(newtask_conv)
+
+    edittask_conv = ConversationHandler(
+        entry_points=[CommandHandler("edittask", edittask_start)],
+        states={
+            EDIT_TASK_NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, edittask_get_new_name)],
+            EDIT_TASK_NEW_NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, edittask_save)]
+        },
+        fallbacks=[]
+    )
+    application.add_handler(edittask_conv)
+
+    deletetask_conv = ConversationHandler(
+        entry_points=[CommandHandler("deletetask", deletetask_start)],
+        states={
+            DELETE_TASK_NAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, deletetask_confirm)],
+            DELETE_TASK_CONFIRM: [MessageHandler(filters.TEXT & ~filters.COMMAND, deletetask_finish)]
+        },
+        fallbacks=[CommandHandler("cancel", cancel)]
+    )
+    application.add_handler(deletetask_conv)
+    application.add_handler(CommandHandler("cancel", cancel))
+
+    application.add_handler(
+        MessageHandler(filters.TEXT & ~filters.COMMAND, handle_free_text)
+    )
+    application.run_polling()
+
+if __name__ == "__main__":
+    main()
+

requirements.txt

@@ -0,0 +1,4 @@
+google-adk==1.7.0
+requests
+python-telegram-bot
+

runner.py

@@ -0,0 +1,52 @@
+import asyncio
+from dotenv import load_dotenv
+from google.adk.runners import InMemoryRunner
+from google.genai import types
+from agent import task_manager_agent
+
+load_dotenv()
+
+runner = InMemoryRunner(
+    agent=task_manager_agent, 
+    app_name="task_manager_app")
+
+async def create_session_async():
+    session = await runner.session_service.create_session(
+        app_name="task_manager_app", user_id="user"
+    )
+    return session
+
+
+async def run_agent(user_id:int, session_id: str, user_message: str): 
+    content = types.Content(role="user", parts=[types.Part.from_text(text=user_message)]) + f"user id : {user_id}"
+    response_text = ""
+    for event in runner.run(
+        user_id=user_id,
+        session_id=session_id,
+        new_message=content,
+    ):
+        
+        if event.content.parts and event.content.parts[0].text:
+            response_text += event.content.parts[0].text + "\n"
+    return response_text.strip()
+# Simple main function to test run_agent interactively
+import sys
+
+def main():
+    import asyncio
+    user_id = "user"  # For testing, you can change this to any string
+    loop = asyncio.get_event_loop()
+    session = loop.run_until_complete(create_session_async())
+    print("ADK Agent Test Console. Type 'exit' to quit.")
+    while True:
+        user_message = input("You: ")
+        if user_message.lower() == "exit":
+            break
+        response = loop.run_until_complete(run_agent(user_id, session.id, user_message))
+        print(f"Agent: {response}\n")
+
+if __name__ == "__main__":
+    main()
+
+
+

tools.py

@@ -0,0 +1,14 @@
+
+from db_utils import add_task_db, list_tasks_db, edit_task_db, delete_task_db
+
+def add_task(user_id: int, task_name: str) -> str:
+    return add_task_db(user_id, task_name)
+
+def list_tasks(user_id: int) -> str:
+    return list_tasks_db(user_id)
+
+def edit_task(user_id: int, old_name: str, new_name: str) -> str:
+    return edit_task_db(user_id, old_name, new_name)
+
+def delete_task(user_id: int, task_name: str) -> str:
+    return delete_task_db(user_id, task_name)