File size: 9,336 Bytes
5374a2d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 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 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 |
"""
Example usage of Telegram tools in EvoAgentX
This example demonstrates how to use the various Telegram tools
for messaging, searching, and file operations.
Prerequisites:
1. Telegram API credentials with the following setup:
- Create a new application at https://my.telegram.org/apps
- Get your API ID and API Hash
- Set up your phone number for authentication
2. Set your credentials as environment variables:
export TELEGRAM_API_ID="your_api_id_here"
export TELEGRAM_API_HASH="your_api_hash_here"
export TELEGRAM_PHONE="your_phone_number_here"
Note: The TelegramToolkit will automatically retrieve the credentials from
the environment variables, making it compatible with AI agents.
"""
import os
from evoagentx.tools import TelegramToolkit
def main():
# Initialize the toolkit - credentials will be automatically retrieved from environment
telegram_toolkit = TelegramToolkit()
# Check if credentials are available
if not telegram_toolkit.telegram_base.api_id or not telegram_toolkit.telegram_base.api_hash:
print("Please set TELEGRAM_API_ID and TELEGRAM_API_HASH environment variables")
print("Get your credentials from: https://my.telegram.org/apps")
return
print("=== Telegram Tools Demo ===\n")
# 1. List recent chats
print("1. Listing Recent Chats")
list_chats_tool = telegram_toolkit.get_tool("list_recent_chats")
result = list_chats_tool(limit=5)
if result["success"]:
print(f"Found {result['chats_count']} recent chats:")
for chat in result["chats"]:
print(f" - {chat['title']} ({chat['type']}) - ID: {chat['id']}")
# Use the first non-system chat for other examples
if result["chats"]:
# Skip system chats (like Telegram user with ID 777000)
for chat in result["chats"]:
if chat["id"] != 777000: # Skip Telegram system user
first_chat_id = chat["id"]
first_chat_title = chat["title"]
break
else:
# If all chats are system chats, use the first one anyway
first_chat_id = result["chats"][0]["id"]
first_chat_title = result["chats"][0]["title"]
else:
print(f"Failed to list chats: {result['error']}")
return
print("\n" + "="*50 + "\n")
# 2. Fetch latest messages
print(f"2. Fetching Latest Messages from '{first_chat_title}'")
fetch_messages_tool = telegram_toolkit.get_tool("fetch_latest_messages")
result = fetch_messages_tool(contact_name=first_chat_title, limit=3)
if result["success"]:
print(f"Found {result['messages_count']} recent messages:")
for msg in result["messages"]:
print(f" - [{msg['date']}] {msg['text'][:50]}...")
else:
print(f"Failed to fetch messages: {result['error']}")
print("\n" + "="*50 + "\n")
# 3. Search messages by keyword
print("3. Searching Messages by Keyword")
search_tool = telegram_toolkit.get_tool("search_messages_by_keyword")
result = search_tool(contact_name=first_chat_title, keyword="Hello", limit=3)
if result["success"]:
print(f"Found {result['matches_count']} messages containing 'hello':")
for msg in result["messages"]:
print(f" - [{msg['date']}] {msg['text'][:50]}...")
else:
print(f"Failed to search messages: {result['error']}")
print("\n" + "="*50 + "\n")
# 4. Find files
print("4. Finding Files in Chat")
find_files_tool = telegram_toolkit.get_tool("find_and_retrieve_file")
result = find_files_tool(contact_name="Telegram", filename_query="Kafka")
if result["success"]:
print(f"Found {result['files_found']} files matching 'Kafka':")
for file_info in result["files"]:
print(f" - {file_info['filename']} ({file_info['file_size']} bytes, {file_info['mime_type']})")
else:
print(f"Failed to find files: {result['error']}")
print("\n" + "="*50 + "\n")
# 5. Send a test message using contact name
print("5. Send Test Message by Contact Name")
send_tool = telegram_toolkit.get_tool("send_message_by_name")
# Try to send a message to a contact (you can modify this name)
test_contact_name = "Vinay Kumar" # This should match a contact in your Telegram
test_message = "Hello! This is a test message from EvoAgentX Telegram tools. 🤖"
print(f"Sending test message to contact: '{test_contact_name}'")
result = send_tool(contact_name=test_contact_name, message_text=test_message)
if result["success"]:
print(f"✅ Message sent successfully!")
print(f" Message ID: {result['message_id']}")
print(f" Sent to: {result['chat']['title']}")
print(f" Message: {result['message_text']}")
print(f" Sent at: {result['sent_at']}")
else:
print(f"❌ Failed to send message: {result['error']}")
if 'clarification_needed' in result:
print(" Available contacts:")
for contact in result['clarification_needed']:
print(f" - {contact}")
print("\n" + "="*50 + "\n")
# 6. Summarize contact messages
print("6. Summarize Contact Messages")
summarize_tool = telegram_toolkit.get_tool("summarize_contact_messages")
# Summarize messages from a contact
test_contact_name = "Telegram" # This should match a contact in your Telegram
print(f"Summarizing recent messages from contact: '{test_contact_name}'")
result = summarize_tool(contact_name=test_contact_name, limit=10)
if result["success"]:
print(f"✅ Message summary generated successfully!")
print(f" Contact: {result['contact']['title']}")
print(f" Messages analyzed: {result['messages_analyzed']}")
print(f" Summary:")
print(f" {result['summary']}")
if result['recent_messages']:
print(f" Recent messages preview:")
for i, msg in enumerate(result['recent_messages'][:3], 1):
direction = "You" if msg['is_outgoing'] else test_contact_name
print(f" {i}. [{msg['date']}] {direction}: {msg['text'][:50]}...")
else:
print(f"❌ Failed to summarize messages: {result['error']}")
if 'clarification_needed' in result:
print(" Available contacts:")
for contact in result['clarification_needed']:
print(f" - {contact}")
print("\n" + "=" * 50)
# Test 7: Download File Tool
print("\n7. Download File Tool")
print("Downloading file from contact:", test_contact_name)
download_tool = telegram_toolkit.get_tool("download_file")
download_result = download_tool(
contact_name=test_contact_name,
filename_query="Kafka",
download_dir="downloads"
)
if download_result["success"]:
print(f"✅ File downloaded successfully!")
print(f" 📁 File: {download_result['filename']}")
print(f" 📍 Path: {download_result['file_path']}")
print(f" 📊 Size: {download_result['file_size']} bytes")
print(f" 📂 Directory: {download_result['download_dir']}")
else:
print(f"❌ Download failed: {download_result['error']}")
if 'clarification_needed' in download_result:
print(" Available contacts:")
for contact in download_result['clarification_needed']:
print(f" - {contact}")
print("\n" + "=" * 50)
# Test 8: Read File Content Tool
print("\n8. Read File Content Tool")
print("Reading file content from contact:", test_contact_name)
read_tool = telegram_toolkit.get_tool("read_file_content")
# Test different content types
content_tests = [
("summary", "Document summary"),
("first_lines", "First 3 lines"),
("last_lines", "Last 3 lines")
]
for content_type, description in content_tests:
print(f"\n 🔍 {description}:")
read_result = read_tool(
contact_name=test_contact_name,
filename_query="Kafka",
content_type=content_type,
lines_count=3
)
if read_result["success"]:
print(f" ✅ {description} extracted successfully!")
print(f" 📄 Content preview:")
content_preview = read_result["content"][:200] + "..." if len(read_result["content"]) > 200 else read_result["content"]
print(f" {content_preview}")
if "file_info" in read_result:
file_info = read_result["file_info"]
print(f" 📊 File info: {file_info}")
else:
print(f" ❌ Failed to read {description}: {read_result['error']}")
print("\n=== Demo Complete ===")
print("All 8 Telegram tools are working correctly!")
print("✅ Core Tools (6): fetch_latest_messages, search_messages_by_keyword, send_message_by_name, list_recent_chats, find_and_retrieve_file, summarize_contact_messages")
print("✅ Enhanced File Tools (2): download_file, read_file_content")
if __name__ == "__main__":
main()
|