Conversation Lifecycle

class MyAgent(ConversimpleAgent):
    async def start(self):
        """Called when agent initializes"""
        await super().start()
        # Agent is now ready for conversations

def on_conversation_started(self, conversation_id: str):
    """Called when user connects"""
    print(f"User connected: {conversation_id}")
    # Initialize conversation-specific state
    self.conversation_data[conversation_id] = {
        "start_time": datetime.now(),
        "context": {}
    }

User Speech → STT → AI → TTS → User Hears
                ↓
            Tool Calls
                ↓
            Your Agent

def on_tool_called(self, tool_call):
    """Called when AI needs to execute a tool"""
    print(f"Executing: {tool_call.tool_name}")

@tool("Look up customer")
def get_customer(self, customer_id: str) -> dict:
    """Your tool executes during active phase"""
    return {"name": "John", "tier": "premium"}

@tool_async("Complex operation")
async def complex_task(self, param: str) -> dict:
    """Long-running tool"""
    await asyncio.sleep(2)  # Simulating API call
    return {"result": "success"}

def on_tool_completed(self, call_id: str, result):
    """Called when tool finishes"""
    print(f"Tool {call_id} completed: {result}")

def on_conversation_ended(self, conversation_id: str):
    """Called when conversation is ending"""
    print(f"Conversation {conversation_id} ending")
    # Save conversation data
    self.save_conversation_log(conversation_id)

def on_conversation_ended(self, conversation_id: str):
    """Final cleanup"""
    # Remove conversation-specific data
    if conversation_id in self.conversation_data:
        data = self.conversation_data.pop(conversation_id)

        # Log metrics
        duration = (datetime.now() - data['start_time']).total_seconds()
        print(f"Conversation lasted {duration:.1f} seconds")

        # Persist important data
        self.save_to_database(conversation_id, data)

from conversimple import ConversimpleAgent, tool

class LifecycleAgent(ConversimpleAgent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.active_conversations = {}

    def on_conversation_started(self, conversation_id: str):
        """Conversation begins"""
        self.active_conversations[conversation_id] = {
            "start_time": datetime.now(),
            "tool_calls": 0,
            "context": {}
        }
        print(f"🎤 Started: {conversation_id}")

    def on_tool_called(self, tool_call):
        """Tool is about to execute"""
        conv_id = tool_call.conversation_id
        self.active_conversations[conv_id]["tool_calls"] += 1
        print(f"🔧 Tool: {tool_call.tool_name}")

    def on_tool_completed(self, call_id: str, result):
        """Tool finished executing"""
        print(f"✅ Completed: {call_id}")

    def on_conversation_ended(self, conversation_id: str):
        """Conversation ends"""
        if conversation_id in self.active_conversations:
            data = self.active_conversations.pop(conversation_id)
            duration = (datetime.now() - data['start_time']).total_seconds()

            print(f"📞 Ended: {conversation_id}")
            print(f"   Duration: {duration:.1f}s")
            print(f"   Tool calls: {data['tool_calls']}")

    def on_error(self, error_type: str, message: str, details: dict):
        """Error occurred"""
        print(f"❌ Error: {error_type} - {message}")
        # Handle error appropriately

class StatefulAgent(ConversimpleAgent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.conversations = {}

    def on_conversation_started(self, conversation_id: str):
        """Initialize conversation state"""
        self.conversations[conversation_id] = {
            "customer_id": None,
            "cart": [],
            "authenticated": False,
            "context": {}
        }

    @tool("Authenticate user")
    def authenticate(self, user_id: str, pin: str) -> dict:
        """Update conversation state"""
        conv_id = self.get_current_conversation_id()
        self.conversations[conv_id]["customer_id"] = user_id
        self.conversations[conv_id]["authenticated"] = True
        return {"authenticated": True}

    @tool("Add to cart")
    def add_to_cart(self, product_id: str) -> dict:
        """Access conversation state"""
        conv_id = self.get_current_conversation_id()
        conv = self.conversations[conv_id]

        if not conv["authenticated"]:
            return {"error": "Please authenticate first"}

        conv["cart"].append(product_id)
        return {"cart_size": len(conv["cart"])}

    def on_conversation_ended(self, conversation_id: str):
        """Clean up conversation state"""
        if conversation_id in self.conversations:
            # Save state before cleanup
            state = self.conversations.pop(conversation_id)
            if state["cart"]:
                # Save abandoned cart
                self.save_cart(state["customer_id"], state["cart"])

import redis
import json

class PersistentAgent(ConversimpleAgent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.redis = redis.Redis(host='localhost', port=6379, db=0)

    @tool("Get customer info")
    def get_customer(self, customer_id: str) -> dict:
        """Load from persistent storage"""
        data = self.redis.get(f"customer:{customer_id}")
        if data:
            return json.loads(data)
        return {"error": "Customer not found"}

    def on_conversation_ended(self, conversation_id: str):
        """Save to persistent storage"""
        conv_data = self.conversations.get(conversation_id)
        if conv_data and conv_data.get("customer_id"):
            # Update customer profile
            customer_id = conv_data["customer_id"]
            key = f"customer:{customer_id}"
            self.redis.set(key, json.dumps(conv_data["context"]))

def on_conversation_ended(self, conversation_id: str):
    """Handle unexpected endings"""
    conv = self.conversations.get(conversation_id)
    if conv and conv.get("in_transaction"):
        # Rollback incomplete transaction
        self.rollback_transaction(conv["transaction_id"])
        print(f"⚠️ Rolled back incomplete transaction")

class MultiConversationAgent(ConversimpleAgent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.conversations = {}
        self.active_count = 0

    def on_conversation_started(self, conversation_id: str):
        self.active_count += 1
        print(f"Active conversations: {self.active_count}")

        if self.active_count > 100:
            print("⚠️ High load - consider scaling")

    def on_conversation_ended(self, conversation_id: str):
        self.active_count -= 1
        print(f"Active conversations: {self.active_count}")

import asyncio
from datetime import datetime, timedelta

class TimeoutAgent(ConversimpleAgent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.last_activity = {}
        self.timeout_task = None

    def on_conversation_started(self, conversation_id: str):
        self.last_activity[conversation_id] = datetime.now()

        # Start monitoring task
        if not self.timeout_task:
            self.timeout_task = asyncio.create_task(self.monitor_timeouts())

    def on_tool_called(self, tool_call):
        """Update activity timestamp"""
        self.last_activity[tool_call.conversation_id] = datetime.now()

    async def monitor_timeouts(self):
        """Check for inactive conversations"""
        while True:
            await asyncio.sleep(30)  # Check every 30 seconds

            now = datetime.now()
            timeout_threshold = timedelta(minutes=5)

            for conv_id, last_active in list(self.last_activity.items()):
                if now - last_active > timeout_threshold:
                    print(f"⏱️ Conversation {conv_id} timed out")
                    # Conversation will end automatically