agent_chat.py•19.1 kB
#!/usr/bin/env python3
"""
Interactive Agent-to-Agent (A2A) Chat Dashboard
Connect your resume-agent with other agents in the NANDA network
"""
import requests
import sys
import json
from datetime import datetime
import time
import os
# ANSI color codes
class Colors:
HEADER = '\033[95m'
BLUE = '\033[94m'
CYAN = '\033[96m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
# Configuration
REGISTRY_URL = "http://registry.chat39.com:6900"
MY_AGENT_ID = "resume-agent"
MY_AGENT_URL = "http://52.73.59.119:6050/a2a"
class AgentRegistry:
def __init__(self):
self.agents = []
self.last_fetch = None
def fetch_agents(self):
"""Fetch all agents from registry"""
try:
response = requests.get(f"{REGISTRY_URL}/list", timeout=10)
if response.status_code == 200:
data = response.json()
self.agents = data.get('agents', [])
self.last_fetch = datetime.now()
return True
else:
print(f"{Colors.RED}❌ Failed to fetch agents: {response.status_code}{Colors.ENDC}")
return False
except Exception as e:
print(f"{Colors.RED}❌ Error: {e}{Colors.ENDC}")
return False
def list_agents(self, filter_text=None, status_filter=None, limit=50):
"""List agents with optional filtering"""
filtered = self.agents
# Filter by status
if status_filter:
filtered = [a for a in filtered if a.get('status', '').lower() == status_filter.lower()]
# Filter by search text
if filter_text:
filter_lower = filter_text.lower()
filtered = [a for a in filtered if
filter_lower in a.get('agent_id', '').lower() or
filter_lower in a.get('name', '').lower() or
filter_lower in str(a.get('expertise', [])).lower() or
filter_lower in str(a.get('specialties', [])).lower()]
return filtered[:limit]
def get_agent(self, agent_id):
"""Get specific agent by ID"""
for agent in self.agents:
if agent.get('agent_id') == agent_id:
# Normalize agent data for consistency
normalized = agent.copy()
# Ensure a2a_endpoint exists
if not normalized.get('a2a_endpoint'):
if normalized.get('agent_url'):
# Add /a2a suffix if not present
url = normalized['agent_url']
normalized['a2a_endpoint'] = f"{url}/a2a" if not url.endswith('/a2a') else url
elif normalized.get('endpoint'):
url = normalized['endpoint']
normalized['a2a_endpoint'] = f"{url}/a2a" if not url.endswith('/a2a') else url
# Ensure name exists
if not normalized.get('name'):
normalized['name'] = f"NANDA Agent {agent_id}"
# Ensure status exists
if not normalized.get('status'):
normalized['status'] = 'unknown'
# Add default expertise if missing
if not normalized.get('expertise') and not normalized.get('specialties'):
if agent_id == 'resume-agent':
normalized['expertise'] = ['resume analysis', 'career information', 'personal assistant']
normalized['description'] = "Personal Resume Agent - Provides information about Vikram Siwach's professional background, skills, and experience"
return normalized
return None
class A2AMessenger:
def __init__(self):
self.conversation_history = []
def send_message(self, target_agent, message_text):
"""Send A2A message from resume-agent to target agent"""
# Get target endpoint
endpoint = target_agent.get('a2a_endpoint') or target_agent.get('endpoint')
if not endpoint:
return None, "No A2A endpoint found for this agent"
# Ensure endpoint has /a2a suffix
if not endpoint.endswith('/a2a'):
endpoint = f"{endpoint}/a2a"
conversation_id = f"resume-to-{target_agent['agent_id']}-{datetime.now().strftime('%Y%m%d-%H%M%S')}"
payload = {
"content": {
"text": message_text,
"type": "text"
},
"role": "agent",
"sender_agent_id": MY_AGENT_ID,
"conversation_id": conversation_id
}
try:
print(f"\n{Colors.CYAN}{'─' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}📤 Sending message to {target_agent['agent_id']}...{Colors.ENDC}")
print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}")
print(f"{Colors.BLUE}From:{Colors.ENDC} {MY_AGENT_ID}")
print(f"{Colors.BLUE}To:{Colors.ENDC} {target_agent['agent_id']}")
print(f"{Colors.BLUE}Msg:{Colors.ENDC} {message_text}")
# Loading animation
for i in range(3):
print(f"\r {'.' * (i+1)}", end='', flush=True)
time.sleep(0.3)
print("\r ", end='')
response = requests.post(endpoint, json=payload, timeout=20)
if response.status_code == 200:
result = response.json()
# Extract response text
response_text = None
if 'parts' in result and len(result['parts']) > 0:
response_text = result['parts'][0].get('text', 'No text in response')
elif 'content' in result:
response_text = result['content'].get('text', 'No text in response')
else:
response_text = str(result)
# Store in history
self.conversation_history.append({
'timestamp': datetime.now(),
'from': MY_AGENT_ID,
'to': target_agent['agent_id'],
'message': message_text,
'response': response_text,
'conv_id': conversation_id
})
return response_text, None
else:
error = f"HTTP {response.status_code}: {response.text[:200]}"
return None, error
except requests.exceptions.Timeout:
return None, "Timeout: Agent took too long to respond"
except requests.exceptions.ConnectionError:
return None, "Connection Error: Could not reach agent endpoint"
except Exception as e:
return None, f"Error: {str(e)}"
def clear_screen():
os.system('clear' if os.name != 'nt' else 'cls')
def display_agent_details(agent):
"""Display detailed information about an agent"""
print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}🤖 Agent Details{Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f" {Colors.BOLD}ID:{Colors.ENDC} {agent.get('agent_id', 'N/A')}")
print(f" {Colors.BOLD}Name:{Colors.ENDC} {agent.get('name', 'N/A')}")
# Status display with colors
status = agent.get('status', 'unknown')
if agent.get('agent_id') == 'resume-agent':
status_display = f"{Colors.GREEN}running (your agent){Colors.ENDC}"
elif status == 'running' or status == 'active':
status_display = f"{Colors.GREEN}{status}{Colors.ENDC}"
elif status == 'unknown':
status_display = f"{Colors.YELLOW}registered{Colors.ENDC}"
else:
status_display = f"{Colors.YELLOW}{status}{Colors.ENDC}"
print(f" {Colors.BOLD}Status:{Colors.ENDC} {status_display}")
print(f" {Colors.BOLD}Endpoint:{Colors.ENDC} {agent.get('a2a_endpoint') or agent.get('endpoint') or agent.get('agent_url', 'N/A')}")
# Expertise and specialties
expertise = agent.get('expertise', agent.get('specialties', []))
if expertise:
print(f" {Colors.BOLD}Expertise:{Colors.ENDC} {', '.join(expertise)}")
# Description
if agent.get('description'):
desc = agent.get('description', 'N/A')
# Wrap description at 60 chars
if len(desc) > 60:
print(f" {Colors.BOLD}Description:{Colors.ENDC}")
words = desc.split()
line = " "
for word in words:
if len(line) + len(word) + 1 <= 74:
line += word + " "
else:
print(line)
line = " " + word + " "
if line.strip():
print(line)
else:
print(f" {Colors.BOLD}Description:{Colors.ENDC} {desc}")
# Last seen
last_seen = agent.get('last_seen', 'N/A')
if last_seen != 'N/A':
print(f" {Colors.BOLD}Last Seen:{Colors.ENDC} {last_seen}")
# Facts URL
if agent.get('facts_url') or agent.get('agentFactsURL'):
facts_url = agent.get('facts_url') or agent.get('agentFactsURL')
print(f" {Colors.BOLD}Facts:{Colors.ENDC} {facts_url}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n")
def display_response(response_text, error=None):
"""Display agent response"""
if error:
print(f"\r{Colors.RED}❌ Failed to get response{Colors.ENDC}")
print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}")
print(f"{Colors.RED}Error:{Colors.ENDC} {error}")
print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}")
else:
print(f"\r{Colors.GREEN}✅ Response received!{Colors.ENDC}")
print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}")
print(f"{Colors.GREEN}{Colors.BOLD}Response:{Colors.ENDC}")
print(f" {response_text}")
print(f"{Colors.CYAN}{'─' * 70}{Colors.ENDC}")
def show_help():
"""Display help information"""
print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}📚 Available Commands{Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f" {Colors.BOLD}/list [filter]{Colors.ENDC} - List agents (optional: filter by name/expertise)")
print(f" {Colors.BOLD}/search <text>{Colors.ENDC} - Search agents by keyword")
print(f" {Colors.BOLD}/info <agent-id>{Colors.ENDC} - Show detailed info about an agent")
print(f" {Colors.BOLD}/chat <agent-id>{Colors.ENDC} - Start chatting with an agent")
print(f" {Colors.BOLD}/history{Colors.ENDC} - Show conversation history")
print(f" {Colors.BOLD}/refresh{Colors.ENDC} - Refresh agent list from registry")
print(f" {Colors.BOLD}/help{Colors.ENDC} - Show this help message")
print(f" {Colors.BOLD}quit{Colors.ENDC} - Exit the dashboard")
print(f"\n{Colors.YELLOW}💡 Quick Examples:{Colors.ENDC}")
print(f" /list - Show all agents")
print(f" /search python - Find Python-related agents")
print(f" /info tech-expert - Get details about tech-expert")
print(f" /chat tech-expert - Chat with tech-expert")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n")
def main():
registry = AgentRegistry()
messenger = A2AMessenger()
current_chat_agent = None
# Welcome screen
clear_screen()
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}{Colors.HEADER}🤖 Resume Agent - A2A Network Chat{Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.GREEN}✓{Colors.ENDC} Your Agent: {Colors.BOLD}{MY_AGENT_ID}{Colors.ENDC}")
print(f"{Colors.GREEN}✓{Colors.ENDC} Endpoint: {Colors.BOLD}{MY_AGENT_URL}{Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n")
# Fetch agents
print(f"{Colors.YELLOW}⏳ Fetching agents from registry...{Colors.ENDC}")
if registry.fetch_agents():
print(f"{Colors.GREEN}✅ Found {len(registry.agents)} agents in the network!{Colors.ENDC}")
print(f"{Colors.YELLOW}💡 Type {Colors.BOLD}/help{Colors.ENDC}{Colors.YELLOW} to see available commands{Colors.ENDC}\n")
else:
print(f"{Colors.RED}❌ Failed to fetch agents. Please check your connection.{Colors.ENDC}\n")
return
# Main loop
while True:
try:
# If in chat mode
if current_chat_agent:
prompt = f"{Colors.BOLD}{Colors.GREEN}💬 [{current_chat_agent['agent_id']}] >{Colors.ENDC} "
else:
prompt = f"{Colors.BOLD}{Colors.BLUE}📝 >{Colors.ENDC} "
user_input = input(prompt).strip()
if not user_input:
continue
# Exit commands
if user_input.lower() in ['quit', 'exit', 'q']:
print(f"\n{Colors.GREEN}👋 Goodbye!{Colors.ENDC}\n")
break
# Help command
elif user_input.lower() == '/help':
show_help()
# List agents
elif user_input.lower().startswith('/list'):
parts = user_input.split(maxsplit=1)
filter_text = parts[1] if len(parts) > 1 else None
agents = registry.list_agents(filter_text=filter_text)
print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}📋 Available Agents ({len(agents)} shown){Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n")
for i, agent in enumerate(agents, 1):
status_color = Colors.GREEN if agent.get('status') == 'running' else Colors.YELLOW
expertise = agent.get('expertise', agent.get('specialties', []))
expertise_str = ', '.join(expertise[:2]) if expertise else 'General'
print(f"{i:3d}. {Colors.BOLD}{agent.get('agent_id', 'unknown')[:30]:30s}{Colors.ENDC} "
f"{status_color}●{Colors.ENDC} {expertise_str[:30]}")
print(f"\n{Colors.YELLOW}💡 Use /info <agent-id> to see details or /chat <agent-id> to start chatting{Colors.ENDC}\n")
# Search agents
elif user_input.lower().startswith('/search'):
parts = user_input.split(maxsplit=1)
if len(parts) < 2:
print(f"{Colors.RED}❌ Usage: /search <keyword>{Colors.ENDC}\n")
continue
agents = registry.list_agents(filter_text=parts[1])
print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}🔍 Search Results: '{parts[1]}' ({len(agents)} found){Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n")
for i, agent in enumerate(agents, 1):
print(f"{i:3d}. {Colors.BOLD}{agent.get('agent_id')}{Colors.ENDC}")
if agent.get('name'):
print(f" {agent.get('name', 'N/A')}")
if agent.get('expertise'):
print(f" {Colors.CYAN}Expertise:{Colors.ENDC} {', '.join(agent.get('expertise', []))}")
print()
# Agent info
elif user_input.lower().startswith('/info'):
parts = user_input.split(maxsplit=1)
if len(parts) < 2:
print(f"{Colors.RED}❌ Usage: /info <agent-id>{Colors.ENDC}\n")
continue
agent = registry.get_agent(parts[1])
if agent:
display_agent_details(agent)
else:
print(f"{Colors.RED}❌ Agent '{parts[1]}' not found{Colors.ENDC}\n")
# Start chat with agent
elif user_input.lower().startswith('/chat'):
parts = user_input.split(maxsplit=1)
if len(parts) < 2:
print(f"{Colors.RED}❌ Usage: /chat <agent-id>{Colors.ENDC}\n")
continue
agent = registry.get_agent(parts[1])
if agent:
current_chat_agent = agent
display_agent_details(agent)
print(f"{Colors.GREEN}✓ Now chatting with {agent['agent_id']}{Colors.ENDC}")
print(f"{Colors.YELLOW}💡 Type your messages directly, or 'back' to return to main menu{Colors.ENDC}\n")
else:
print(f"{Colors.RED}❌ Agent '{parts[1]}' not found{Colors.ENDC}\n")
# Exit chat mode
elif user_input.lower() in ['back', '/back', 'exit chat']:
if current_chat_agent:
print(f"{Colors.YELLOW}← Exited chat with {current_chat_agent['agent_id']}{Colors.ENDC}\n")
current_chat_agent = None
else:
print(f"{Colors.YELLOW}Not in chat mode{Colors.ENDC}\n")
# Show history
elif user_input.lower() == '/history':
if not messenger.conversation_history:
print(f"{Colors.YELLOW}No conversation history yet{Colors.ENDC}\n")
else:
print(f"\n{Colors.CYAN}{'═' * 70}{Colors.ENDC}")
print(f"{Colors.BOLD}💬 Conversation History ({len(messenger.conversation_history)}){Colors.ENDC}")
print(f"{Colors.CYAN}{'═' * 70}{Colors.ENDC}\n")
for i, conv in enumerate(messenger.conversation_history[-10:], 1):
timestamp = conv['timestamp'].strftime('%H:%M:%S')
print(f"{Colors.BOLD}[{i}] {timestamp}{Colors.ENDC}")
print(f" {Colors.BLUE}→{Colors.ENDC} {conv['to']}: {conv['message'][:50]}...")
print(f" {Colors.GREEN}←{Colors.ENDC} {conv['response'][:50]}...")
print()
# Refresh registry
elif user_input.lower() == '/refresh':
print(f"{Colors.YELLOW}⏳ Refreshing agent list...{Colors.ENDC}")
if registry.fetch_agents():
print(f"{Colors.GREEN}✅ Updated! Found {len(registry.agents)} agents{Colors.ENDC}\n")
else:
print(f"{Colors.RED}❌ Failed to refresh{Colors.ENDC}\n")
# Send message (if in chat mode)
elif current_chat_agent:
response_text, error = messenger.send_message(current_chat_agent, user_input)
display_response(response_text, error)
# Unknown command
else:
print(f"{Colors.RED}❌ Unknown command. Type /help for available commands{Colors.ENDC}\n")
except KeyboardInterrupt:
print(f"\n\n{Colors.GREEN}👋 Goodbye!{Colors.ENDC}\n")
break
except EOFError:
break
if __name__ == "__main__":
main()