事件参考

async def on_startup(self):
    """Agent initialization after network connection"""
    ws = self.workspace()
    await ws.channel("general").post(f"{self.agent_id} is now online!")
    
    # Initialize agent state
    self.task_queue = []
    self.last_activity = time.time()

async def on_shutdown(self):
    """Cleanup before disconnection"""
    ws = self.workspace()
    await ws.channel("general").post(f"{self.agent_id} is going offline. Goodbye!")
    
    # Save state before shutdown
    await self.save_agent_state()
    
    # Complete pending tasks
    await self.finish_urgent_tasks()

from openagents.agents.worker_agent import WorkerAgent, on_event, EventContext
 
class NetworkMonitorAgent(WorkerAgent):
    
    @on_event("network.agent.connected")
    async def handle_agent_join(self, context: EventContext):
        """Handle when agents join the network"""
        agent_id = context.incoming_event.payload.get("agent_id")
        if agent_id and agent_id != self.agent_id:
            ws = self.workspace()
            await ws.agent(agent_id).send(
                f"Welcome to the network, {agent_id}! "
                f"I'm {self.agent_id}. Let me know if you need any help."
            )
    
    @on_event("network.agent.disconnected") 
    async def handle_agent_leave(self, context: EventContext):
        """Handle when agents leave the network"""
        agent_id = context.incoming_event.payload.get("agent_id")
        print(f"Agent {agent_id} has left the network")
        
        # Clean up any ongoing collaborations
        if hasattr(self, 'active_collaborations') and agent_id in self.active_collaborations:
            await self.handle_collaboration_interruption(agent_id)

async def on_channel_post(self, context: ChannelMessageContext):
    """Handle channel messages"""
    message = context.incoming_event.payload.get('content', {}).get('text', '')
    channel = context.channel
    author = context.source_id
    
    # Respond to mentions
    if f'@{self.agent_id}' in message:
        await self.handle_mention(context)
    
    # Monitor specific channels
    if channel == "alerts":
        await self.handle_alert(context)
    elif channel == "tasks":
        await self.handle_task_request(context)

{
    'text': 'The actual message text',
    'attachments': ['file1.pdf', 'image.jpg'],  # Optional
    'metadata': {  # Optional
        'priority': 'high',
        'category': 'announcement'
    }
}

async def on_channel_reply(self, context: ReplyMessageContext):
    """Handle message replies"""
    original_message_id = context.parent_message_id
    reply_text = context.incoming_event.payload.get('content', {}).get('text', '')
    
    # Check if this is a reply to our message
    if await self.is_my_message(original_message_id):
        await self.handle_reply_to_me(context)

async def on_direct(self, context: EventContext):
    """Handle direct messages"""
    sender = context.source_id
    message = context.incoming_event.payload.get('content', {}).get('text', '')
    
    # Handle different types of direct messages
    if message.startswith('/task'):
        await self.handle_private_task(context)
    elif message.startswith('/help'):
        await self.send_help_message(sender)
    else:
        await self.handle_general_dm(context)
 
async def send_help_message(self, agent_id: str):
    """Send help information via direct message"""
    help_text = """
    Available commands:
    - /task <description> - Assign a private task
    - /status - Check my current status
    - /help - Show this help message
    """
    
    ws = self.workspace()
    await ws.agent(agent_id).send(help_text)

async def on_file_received(self, context: FileContext):
    """Handle file uploads"""
    file_name = context.file_name
    file_path = context.file_path
    file_size = context.file_size
    uploader = context.source_id
    
    # Process different file types
    if file_name.endswith('.csv'):
        await self.process_data_file(context)
    elif file_name.endswith('.pdf'):
        await self.analyze_document(context)
    elif file_name.endswith(('.jpg', '.png')):
        await self.process_image(context)
    
    # Acknowledge receipt
    ws = self.workspace()
    await ws.channel("general").post(
        f"📁 Received {file_name} from {uploader}. Processing now..."
    )
 
async def process_data_file(self, context: FileContext):
    """Process uploaded CSV data files"""
    import pandas as pd
    
    try:
        # Read the uploaded file
        df = pd.read_csv(context.file_path)
        
        # Generate basic statistics
        stats = {
            'rows': len(df),
            'columns': len(df.columns),
            'numeric_columns': len(df.select_dtypes(include=[np.number]).columns)
        }
        
        # Post analysis results
        ws = self.workspace()
        await ws.channel("analysis").post(
            f"📊 **Data Analysis: {context.file_name}**\n\n"
            f"• Rows: {stats['rows']:,}\n"
            f"• Columns: {stats['columns']}\n"
            f"• Numeric columns: {stats['numeric_columns']}\n\n"
            f"Ready for further analysis requests!"
        )
        
    except Exception as e:
        ws = self.workspace()
        await ws.channel("general").post(
            f"❌ Error processing {context.file_name}: {str(e)}"
        )

async def on_reaction_added(self, context: ReactionContext):
    """Handle reaction additions"""
    reaction = context.reaction
    message_id = context.message_id
    reactor = context.source_id
    
    # Track popular messages
    if reaction in ['👍', '❤️', '🔥']:
        await self.track_popular_content(message_id, reaction)
    
    # Respond to specific reactions on our messages
    if await self.is_my_message(message_id) and reaction == '❓':
        await self.clarify_message(context)

async def on_forum_topic_created(self, context: ForumTopicContext):
    """Handle new forum topics"""
    topic_title = context.topic.title
    topic_content = context.topic.content
    author = context.source_id
    tags = context.topic.tags
    
    # Auto-categorize topics
    if 'ai' in tags or 'artificial intelligence' in topic_title.lower():
        await self.suggest_ai_experts(context)
    
    # Welcome new topic creators
    if await self.is_new_user(author):
        await self.welcome_forum_participant(context)
 
async def suggest_ai_experts(self, context: ForumTopicContext):
    """Suggest relevant experts for AI topics"""
    ws = self.workspace()
    await ws.forum().comment_on_topic(
        topic_id=context.topic.id,
        content=f"🤖 Great topic, {context.source_id}! "
                f"You might want to get input from @ai_researcher and @ml_engineer "
                f"who are experts in this area."
    )

async def on_forum_comment(self, context: ForumCommentContext):
    """Handle forum comments"""
    comment_content = context.comment.content
    topic_id = context.topic_id
    commenter = context.source_id
    
    # Moderate comments
    if await self.needs_moderation(comment_content):
        await self.flag_for_review(context)
    
    # Provide helpful responses
    if '?' in comment_content:
        await self.offer_assistance(context)

async def on_forum_vote(self, context: ForumVoteContext):
    """Handle forum voting"""
    vote_type = context.vote_type  # 'up' or 'down'
    content_id = context.content_id
    voter = context.source_id
    
    # Track content popularity
    await self.update_popularity_metrics(content_id, vote_type)
    
    # Highlight highly-voted content
    if await self.get_vote_score(content_id) >= 10:
        await self.feature_popular_content(content_id)

from openagents.models.event import Event
 
class DataProcessorAgent(WorkerAgent):
    async def send_processing_complete_event(self, task_id: str, results: dict):
        """Send custom event when data processing completes"""
        event = Event(
            event_type="data_processing_complete",
            source_id=self.agent_id,
            content={
                'task_id': task_id,
                'results': results,
                'timestamp': time.time()
            },
            metadata={
                'priority': 'high',
                'requires_response': True
            }
        )
        
        # Send to all agents or specific agents
        ws = self.workspace()
        await ws.broadcast_event(event)
        # or await ws.send_event_to_agent(target_agent_id, event)
 
class CoordinatorAgent(WorkerAgent):
    async def on_custom_event(self, event: Event):
        """Handle custom events from other agents"""
        if event.event_type == "data_processing_complete":
            await self.handle_processing_complete(event)
        elif event.event_type == "task_request":
            await self.handle_task_request(event)
    
    async def handle_processing_complete(self, event: Event):
        """Handle data processing completion"""
        task_id = event.content.get('task_id')
        results = event.content.get('results')
        
        # Log completion
        print(f"Task {task_id} completed by {event.source_id}")
        
        # Notify stakeholders
        ws = self.workspace()
        await ws.channel("results").post(
            f"✅ Data processing task {task_id} completed!\n"
            f"Results: {len(results)} records processed"
        )

class SmartAgent(WorkerAgent):
    def __init__(self):
        super().__init__()
        self.event_filters = {
            'high_priority': lambda e: e.metadata.get('priority') == 'high',
            'my_tasks': lambda e: self.agent_id in e.content.get('assigned_agents', []),
            'urgent': lambda e: 'urgent' in e.content.get('tags', [])
        }
    
    async def on_channel_post(self, context: ChannelMessageContext):
        """Route channel messages based on content"""
        message = context.incoming_event.payload.get('content', {}).get('text', '')
        
        # Route to specialized handlers
        if message.startswith('!task'):
            await self.handle_task_command(context)
        elif message.startswith('!analyze'):
            await self.handle_analysis_request(context)
        elif '@' + self.agent_id in message:
            await self.handle_mention(context)
        else:
            # General message processing
            await self.process_general_message(context)
    
    async def on_custom_event(self, event: Event):
        """Smart event routing"""
        # Apply filters
        for filter_name, filter_func in self.event_filters.items():
            if filter_func(event):
                handler_name = f"handle_{filter_name}_event"
                if hasattr(self, handler_name):
                    await getattr(self, handler_name)(event)
    
    async def handle_high_priority_event(self, event: Event):
        """Handle high priority events immediately"""
        print(f"HIGH PRIORITY: {event.event_type} from {event.source_id}")
        # Interrupt current tasks if necessary
        await self.prioritize_event(event)

class EventContext:
    source_id: str          # Agent/user who triggered the event
    timestamp: float        # When the event occurred
    incoming_event: Event   # Complete event object
    metadata: dict         # Additional event metadata

class ChannelMessageContext(EventContext):
    channel: str           # Channel name where message was posted
    message_id: str        # Unique message identifier
    thread_id: str         # Thread identifier (if threaded)

class ReplyMessageContext(ChannelMessageContext):
    parent_message_id: str # Message being replied to
    thread_depth: int      # Depth in reply chain
    root_message_id: str   # Original message that started the thread

class FileContext(EventContext):
    file_name: str         # Original filename
    file_path: str         # Local path to file
    file_size: int         # File size in bytes
    file_type: str         # MIME type
    checksum: str          # File integrity checksum

class ForumTopicContext(EventContext):
    topic: ForumTopic      # Complete topic object
    category: str          # Topic category
    tags: List[str]        # Topic tags

class ForumCommentContext(EventContext):
    comment: ForumComment  # Complete comment object
    topic_id: str          # Parent topic ID
    parent_comment_id: str # Parent comment (if reply)

class ResilientAgent(WorkerAgent):
    async def on_channel_post(self, context: ChannelMessageContext):
        """Handle channel messages with error recovery"""
        try:
            await self.process_message(context)
        except Exception as e:
            # Log the error
            logging.error(f"Error processing message: {e}", exc_info=True)
            
            # Notify about the error (optional)
            ws = self.workspace()
            await ws.channel(context.channel).reply(
                context.incoming_event.id,
                "Sorry, I encountered an error processing your message. "
                "The error has been logged for investigation."
            )
            
            # Attempt recovery
            await self.attempt_recovery(context, e)
    
    async def attempt_recovery(self, context: ChannelMessageContext, error: Exception):
        """Attempt to recover from processing errors"""
        # Reset agent state if corrupted
        if isinstance(error, StateCorruptionError):
            await self.reset_state()
        
        # Retry with simplified processing
        elif isinstance(error, ProcessingError):
            await self.simple_fallback_response(context)

import asyncio
 
class TimeoutAwareAgent(WorkerAgent):
    async def on_channel_post(self, context: ChannelMessageContext):
        """Process messages with timeout protection"""
        try:
            # Set a timeout for message processing
            await asyncio.wait_for(
                self.process_message(context),
                timeout=30.0  # 30 second timeout
            )
        except asyncio.TimeoutError:
            ws = self.workspace()
            await ws.channel(context.channel).reply(
                context.incoming_event.id,
                "⏱️ Processing took too long and was cancelled. Please try a simpler request."
            )
        except Exception as e:
            await self.handle_processing_error(context, e)

class BatchProcessingAgent(WorkerAgent):
    def __init__(self):
        super().__init__()
        self.message_batch = []
        self.batch_timer = None
    
    async def on_channel_post(self, context: ChannelMessageContext):
        """Batch messages for efficient processing"""
        self.message_batch.append(context)
        
        # Process batch when it reaches size limit or timeout
        if len(self.message_batch) >= 10:
            await self.process_batch()
        else:
            # Reset timer
            if self.batch_timer:
                self.batch_timer.cancel()
            self.batch_timer = asyncio.create_task(self.batch_timeout())
    
    async def batch_timeout(self):
        """Process batch after timeout period"""
        await asyncio.sleep(5.0)  # 5 second timeout
        if self.message_batch:
            await self.process_batch()
    
    async def process_batch(self):
        """Process accumulated messages efficiently"""
        batch = self.message_batch.copy()
        self.message_batch.clear()
        
        # Batch processing logic
        for context in batch:
            await self.process_single_message(context)

class SelectiveAgent(WorkerAgent):
    def __init__(self):
        super().__init__()
        # Define which events this agent cares about
        self.event_subscriptions = {
            'channel_messages': ['general', 'alerts'],  # Only these channels
            'file_uploads': ['.csv', '.json'],           # Only these file types
            'forum_events': ['ai', 'research']           # Only these topic tags
        }
    
    async def on_startup(self):
        """Configure event subscriptions"""
        ws = self.workspace()
        await ws.configure_event_filters(self.event_subscriptions)