The 80% of agents no one shows you

# Python
pip install xpander-sdk

# Node.js
npm install @xpander-ai/sdk

# CLI (for agent creation)
npm install -g xpander-cli

# Install dependencies (Python example)
pip install -r requirements.txt

# Use xpander-cli to scaffold a new agent template
xpander login  
xpander agent new
    

# Real implementation following your working projects
import os
import asyncio
from agno.agent import Agent
from agno.models.anthropic import Claude  # Sonnet 4
from agno.models.openai import OpenAI     # OpenAI models
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory
from agno.storage.sqlite import SqliteStorage
from agno.tools.thinking import ThinkingTools
from agno.tools import tool
from xpander_utils.sdk.adapters import AgnoAdapter
from xpander_utils.events import XpanderEventListener, AgentExecutionResult, AgentExecution
from xpander_sdk import LLMTokens, Tokens
from dotenv import load_dotenv
import json
from pathlib import Path

load_dotenv()

# Load xpander configuration (following your project pattern)
CFG_PATH = Path("xpander_config.json")
xpander_cfg = json.loads(CFG_PATH.read_text()) if CFG_PATH.exists() else {
    "organization_id": "your_org_id",
    "api_key": "your_xpander_api_key",
    "agent_id": "your_customer_service_agent_id"
}

# Customer Service Tools (following your @tool patterns)
@tool(
    name="CustomerSupportTool",
    description="Handle customer support requests, search knowledge base, and create tickets",
    show_result=True,
    stop_after_tool_call=True
)
async def handle_customer_request(query: str, customer_id: str, priority: str = "medium"):
    """Handle customer support requests with knowledge base search"""
    
    # Search knowledge base
    knowledge_results = {
        "articles": [
            {"title": "Account Setup Guide", "relevance": 0.9},
            {"title": "Billing FAQ", "relevance": 0.8},
            {"title": "Technical Support", "relevance": 0.7}
        ],
        "total_found": 3
    }
    
    # Create support ticket
    ticket_id = f"TICKET-{customer_id}-{hash(query) % 10000}"
    
    response = {
        "ticket_id": ticket_id,
        "knowledge_results": knowledge_results,
        "suggested_actions": ["Check account settings", "Verify billing info"],
        "escalation_needed": priority == "high",
        "estimated_resolution": "2-4 hours" if priority != "high" else "1 hour"
    }
    
    print(f"Created support ticket: {ticket_id}")
    return response

# Customer Service Orchestrator (following your project structure)
class CustomerServiceOrchestrator:
    def __init__(self, agent_backend: AgnoAdapter):
        self.agent_backend = agent_backend
        self.agent = None
        self.tools = [
            ThinkingTools(add_instructions=True),
            handle_customer_request
        ]
        
        # Multiple model options - choose one:
        
        # Option 1: Sonnet 4 (Anthropic)
        self.model = Claude(
            id="claude-3-5-sonnet-20241022",
            api_key=os.getenv("ANTHROPIC_API_KEY")
        )
        
        # Option 3: OpenAI
        # self.model = OpenAI(
        #     id="gpt-4o",
        #     api_key=os.getenv("OPENAI_API_KEY")
        # )
        
        # Memory system (real implementation from your projects)
        self.memory = Memory(
            model=self.model,
            db=SqliteMemoryDb(table_name="customer_memories", db_file="customer_support.db")
        )
        self.storage = SqliteStorage(table_name="support_sessions", db_file="customer_support.db")

    async def run(self, message, user_id, session_id, cli=False):
        if self.agent is None:
            self.agent = Agent(
                model=self.model,
                tools=self.tools,
                memory=self.memory,
                enable_agentic_memory=True,
                enable_user_memories=True,
                storage=self.storage,
                add_history_to_messages=True,
                num_history_runs=3,
                instructions=[
                    "You are a customer service agent.",
                    "Always be helpful, professional, and empathetic.",
                    "Search knowledge base before creating new tickets.",
                    "Escalate complex issues when needed.",
                    f"User: {user_id}",
                    f"Session: {session_id}"
                ]
            )
        
        if cli:
            return await self.agent.aprint_response(
                message,
                user_id=user_id,
                session_id=session_id,
                stream=True,
                stream_intermediate_steps=True
            )
        else:
            return await self.agent.arun(
                message,
                user_id=user_id,
                session_id=session_id
            )

# Initialize xpander backend (following your pattern)
xpander_backend = AgnoAdapter(
    agent_id=xpander_cfg["agent_id"],
    api_key=xpander_cfg["api_key"]
)

# Create orchestrator instance
customer_service_orchestrator = CustomerServiceOrchestrator(xpander_backend)
    

# Real memory implementation following your working projects
import sqlite3
from datetime import datetime
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.memory.v2.memory import Memory
from agno.storage.sqlite import SqliteStorage

# Database initialization (following your pr_review.py pattern)
DB_PATH = "customer_support.db"

def init_customer_db():
    """Initialize customer support database"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # Customer interactions table
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS customer_interactions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id TEXT,
            session_id TEXT,
            query TEXT,
            response TEXT,
            ticket_id TEXT,
            priority TEXT,
            resolved BOOLEAN DEFAULT FALSE,
            created_at TEXT
        )
    ''')
    
    # Customer preferences table
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS customer_preferences (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id TEXT UNIQUE,
            communication_style TEXT,
            preferred_language TEXT,
            technical_level TEXT,
            updated_at TEXT
        )
    ''')
    
    conn.commit()
    conn.close()

# Initialize database
init_customer_db()

# Memory and storage setup (real pattern from your projects)
class CustomerServiceMemory:
    """Real memory management using Agno framework"""
    
    def __init__(self, model):
        self.memory = Memory(
            model=model,
            db=SqliteMemoryDb(
                table_name="customer_memories", 
                db_file=DB_PATH
            )
        )
        self.storage = SqliteStorage(
            table_name="customer_sessions", 
            db_file=DB_PATH
        )
    
    def store_customer_interaction(self, customer_id: str, query: str, 
                                 response: str, ticket_id: str = None):
        """Store customer interaction for future reference"""
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO customer_interactions 
            (customer_id, query, response, ticket_id, created_at)
            VALUES (?, ?, ?, ?, ?)
        ''', (customer_id, query, response, ticket_id, datetime.now().isoformat()))
        
        conn.commit()
        conn.close()
    
    def get_customer_history(self, customer_id: str, limit: int = 10):
        """Retrieve recent customer interactions"""
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT query, response, ticket_id, created_at
            FROM customer_interactions
            WHERE customer_id = ?
            ORDER BY created_at DESC
            LIMIT ?
        ''', (customer_id, limit))
        
        history = cursor.fetchall()
        conn.close()
        
        return [
            {
                "query": row[0],
                "response": row[1], 
                "ticket_id": row[2],
                "timestamp": row[3]
            }
            for row in history
        ]
    
    def update_customer_preferences(self, customer_id: str, preferences: dict):
        """Store customer communication preferences"""
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT OR REPLACE INTO customer_preferences
            (customer_id, communication_style, preferred_language, 
             technical_level, updated_at)
            VALUES (?, ?, ?, ?, ?)
        ''', (
            customer_id,
            preferences.get("communication_style", "friendly"),
            preferences.get("preferred_language", "english"),
            preferences.get("technical_level", "intermediate"),
            datetime.now().isoformat()
        ))
        
        conn.commit()
        conn.close()
    
    def get_customer_preferences(self, customer_id: str):
        """Retrieve customer preferences for personalization"""
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        
        cursor.execute('''
            SELECT communication_style, preferred_language, technical_level
            FROM customer_preferences
            WHERE customer_id = ?
        ''', (customer_id,))
        
        result = cursor.fetchone()
        conn.close()
        
        if result:
            return {
                "communication_style": result[0],
                "preferred_language": result[1],
                "technical_level": result[2]
            }
        return None

# Usage example following your project patterns
def setup_customer_memory():
    """Setup memory system with different model options"""
    
    # Choose your model (multiple options like your projects)
    from agno.models.anthropic import Claude
    # from agno.models.openai import OpenAI
    
    model = Claude(
        id="claude-3-5-sonnet-20241022",
        api_key=os.getenv("ANTHROPIC_API_KEY")
    )
    
    # Initialize memory system
    memory_system = CustomerServiceMemory(model)
    
    # Example: Store customer interaction
    memory_system.store_customer_interaction(
        customer_id="customer_123",
        query="How do I reset my password?",
        response="I've created a password reset ticket for you.",
        ticket_id="TICKET-customer_123-1234"
    )
    
    # Example: Update customer preferences
    memory_system.update_customer_preferences("customer_123", {
        "communication_style": "technical",
        "preferred_language": "english", 
        "technical_level": "advanced"
    })
    
    # Example: Retrieve customer context
    history = memory_system.get_customer_history("customer_123")
    preferences = memory_system.get_customer_preferences("customer_123")
    
    print(f"Customer history: {len(history)} interactions")
    print(f"Customer preferences: {preferences}")
    
    return memory_system

# Initialize memory system
customer_memory = setup_customer_memory()
    

# Real tool implementation following your working projects
import requests
import json
import sqlite3
from datetime import datetime
from agno.tools import tool
from dotenv import load_dotenv
import os

load_dotenv()

# Customer Support Tools (following your @tool decorator patterns)
@tool(
    name="KnowledgeSearchTool",
    description="Search internal knowledge base for customer support articles and solutions",
    show_result=True,
    stop_after_tool_call=False
)
async def search_knowledge_base(query: str, category: str = "general"):
    """Search knowledge base for relevant support information"""
    
    # Simulate knowledge base (in production, use vector search)
    knowledge_base = {
        "account": [
            {"title": "Password Reset Guide", "content": "Steps to reset your password", "relevance": 0.9},
            {"title": "Account Verification", "content": "How to verify your account", "relevance": 0.8},
            {"title": "Profile Settings", "content": "Managing your profile", "relevance": 0.7}
        ],
        "billing": [
            {"title": "Payment Methods", "content": "Accepted payment methods", "relevance": 0.9},
            {"title": "Billing Cycles", "content": "Understanding billing periods", "relevance": 0.8},
            {"title": "Refund Policy", "content": "How to request refunds", "relevance": 0.7}
        ],
        "technical": [
            {"title": "API Integration", "content": "How to integrate our API", "relevance": 0.9},
            {"title": "Troubleshooting", "content": "Common technical issues", "relevance": 0.8},
            {"title": "Performance Optimization", "content": "Tips for better performance", "relevance": 0.7}
        ]
    }
    
    results = []
    search_categories = [category] if category != "general" else knowledge_base.keys()
    
    for cat in search_categories:
        if cat in knowledge_base:
            for article in knowledge_base[cat]:
                if query.lower() in article["title"].lower() or query.lower() in article["content"].lower():
                    results.append({
                        "category": cat,
                        "title": article["title"],
                        "content": article["content"],
                        "relevance": article["relevance"]
                    })
    
    results.sort(key=lambda x: x["relevance"], reverse=True)
    
    return {
        "query": query,
        "category": category,
        "results": results[:5],
        "total_found": len(results)
    }

@tool(
    name="TicketManagementTool", 
    description="Create, update, and manage customer support tickets",
    show_result=True,
    stop_after_tool_call=False
)
async def manage_support_ticket(action: str, customer_id: str, issue_description: str = None, 
                               ticket_id: str = None, priority: str = "medium", status: str = "open"):
    """Manage customer support tickets - create, update, or close"""
    
    if action == "create":
        ticket_id = f"TICKET-{customer_id}-{hash(issue_description or '') % 10000}"
        conn = sqlite3.connect("customer_support.db")
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO customer_interactions 
            (customer_id, query, ticket_id, priority, created_at)
            VALUES (?, ?, ?, ?, ?)
        ''', (customer_id, issue_description, ticket_id, priority, datetime.now().isoformat()))
        
        conn.commit()
        conn.close()
        
        return {
            "action": "created",
            "ticket_id": ticket_id,
            "customer_id": customer_id,
            "priority": priority,
            "estimated_resolution": "2-4 hours" if priority != "high" else "1 hour",
            "status": "open"
        }
    
    elif action == "update":
        conn = sqlite3.connect("customer_support.db")
        cursor = conn.cursor()
        
        cursor.execute('''
            UPDATE customer_interactions 
            SET resolved = ?
            WHERE ticket_id = ?
        ''', (status == "closed", ticket_id))
        
        conn.commit()
        conn.close()
        
        return {
            "action": "updated",
            "ticket_id": ticket_id,
            "status": status
        }

@tool(
    name="ExternalAPITool",
    description="Call external APIs for customer verification, billing, or integrations",
    show_result=True,
    stop_after_tool_call=False
)
async def call_external_api(service: str, endpoint: str, customer_id: str, **kwargs):
    """Call external services (following your Slack integration pattern)"""
    
    if service == "billing":
        return {
            "service": "billing",
            "customer_id": customer_id,
            "account_status": "active",
            "current_plan": "premium",
            "billing_email": f"{customer_id}@example.com",
            "next_billing_date": "2024-02-01"
        }
    
    elif service == "notification" and endpoint == "slack":
        slack_token = os.getenv("SLACK_BOT_TOKEN")
        slack_channel = os.getenv("SLACK_CHANNEL_ID")
        
        if slack_token and slack_channel:
            message = kwargs.get("message", "Customer support notification")
            headers = {
                "Authorization": f"Bearer {slack_token}",
                "Content-Type": "application/json"
            }
            payload = {
                "channel": slack_channel,
                "text": f"🎫 Customer Support Alert: {message}",
                "unfurl_links": False
            }
            
            try:
                response = requests.post(
                    "https://slack.com/api/chat.postMessage",
                    headers=headers,
                    data=json.dumps(payload)
                )
                result = response.json()
                return {"service": "notification", "platform": "slack", "sent": result.get("ok", False)}
            except Exception as e:
                return {"service": "notification", "error": str(e)}
        
        return {"service": "notification", "error": "Slack credentials not configured"}
    
    return {"service": service, "error": "Service not supported"}

# Tool usage example (following your orchestrator pattern)
async def test_customer_tools():
    """Test customer service tools"""
    
    kb_result = await search_knowledge_base("password reset", "account")
    print(f"Knowledge base search: {kb_result}")
    
    ticket_result = await manage_support_ticket(
        action="create",
        customer_id="customer_123",
        issue_description="Cannot access my account",
        priority="high"
    )
    print(f"Ticket created: {ticket_result}")

# Uncomment to test tools
# asyncio.run(test_customer_tools())
    

# main.py - CLI interface (following your main.py pattern)
import asyncio
import json
from pathlib import Path

async def main():
    """Main CLI interface for customer service agent"""
    
    # Load configuration (following your pattern)
    CFG_PATH = Path("xpander_config.json")
    xpander_cfg = json.loads(CFG_PATH.read_text()) if CFG_PATH.exists() else {
        "organization_id": "your_org_id",
        "api_key": "your_xpander_api_key",
        "agent_id": "your_customer_service_agent_id"
    }
    
    # Initialize xpander backend
    xpander_backend = AgnoAdapter(
        agent_id=xpander_cfg["agent_id"],
        api_key=xpander_cfg["api_key"]
    )
    
    # Create orchestrator
    orchestrator = CustomerServiceOrchestrator(xpander_backend)
    
    print("🎫 Customer Service Agent Started")
    print("Type 'exit' or 'quit' to stop")
    
    while True:
        message = input("\nCustomer: ")
        if message.lower() in ["exit", "quit"]:
            break
        
        try:
            response = await orchestrator.run(
                message=message,
                user_id="customer123",
                session_id="support_session456",
                cli=True
            )
        except Exception as e:
            print(f"Error: {str(e)}")

if __name__ == "__main__":
    asyncio.run(main())
    

# xpander_handler.py - Event processing (following your xpander_handler.py pattern)
import json
import asyncio
from xpander_utils.events import XpanderEventListener, AgentExecutionResult, AgentExecution
from xpander_utils.sdk.adapters import AgnoAdapter
from pathlib import Path
from dotenv import load_dotenv
from xpander_sdk import LLMTokens, Tokens

load_dotenv()

# Load configuration
CFG_PATH = Path("xpander_config.json")
if not CFG_PATH.exists():
    raise FileNotFoundError("Missing xpander_config.json")

xpander_cfg: dict = json.loads(CFG_PATH.read_text())

# Initialize xpander backend (following your pattern)
xpander_backend: AgnoAdapter = asyncio.run(
    asyncio.to_thread(AgnoAdapter, agent_id=xpander_cfg["agent_id"], api_key=xpander_cfg["api_key"])
)

# Create orchestrator
customer_service_orchestrator = CustomerServiceOrchestrator(xpander_backend)

# Event handler (following your pattern exactly)
async def on_execution_request(execution_task: AgentExecution) -> AgentExecutionResult:
    """
    Handle customer service requests from Xpander UI
    
    Args:
        execution_task (AgentExecution): Object containing execution metadata and input.

    Returns:
        AgentExecutionResult: Object describing the output of the execution.
    """
    
    try:
        # Initialize the agent with task (following your pattern)
        await asyncio.to_thread(xpander_backend.agent.init_task, execution=execution_task.model_dump())
        
        # Format customer request (following your pattern)
        customer_request = (
            f"Customer inquiry: {execution_task.input.text}\n"
            f"Customer: {execution_task.input.user.first_name} "
            f"{execution_task.input.user.last_name}\n"
            f"Email: {execution_task.input.user.email}"
        )
        
        # Process with orchestrator
        agent_response = await customer_service_orchestrator.run(
            customer_request, 
            execution_task.input.user.id, 
            execution_task.memory_thread_id
        )
        
        # Report metrics (following your pattern exactly)
        metrics = agent_response.metrics
        llm_tokens = LLMTokens(
            completion_tokens=sum(metrics['completion_tokens']),
            prompt_tokens=sum(metrics['prompt_tokens']),
            total_tokens=sum(metrics['total_tokens']),
        )
        xpander_backend.agent.report_execution_metrics(
            llm_tokens=Tokens(worker=llm_tokens), 
            ai_model=agent_response.model, 
            source_node_type="agno"
        )
        
    except Exception as e:
        return AgentExecutionResult(result=str(e), is_success=False)
    
    return AgentExecutionResult(result=agent_response.content, is_success=True)

# Register event listener (following your pattern)
listener = XpanderEventListener(**xpander_cfg)
listener.register(on_execution_request=on_execution_request)

print("🎫 Customer Service Agent listening for events...")
print("Access the agent from your Xpander UI")
    

// xpander_config.json (following your project structure)
{
  "organization_id": "your_organization_id",
  "api_key": "your_xpander_api_key",
  "agent_id": "your_customer_service_agent_id"
}
    

# .env file (following your project pattern)
# AI Model APIs (choose one or multiple)
ANTHROPIC_API_KEY=your_anthropic_key  # For Sonnet 4
OPENAI_API_KEY=your_openai_key        # For GPT models

# Slack integration (optional)
SLACK_BOT_TOKEN=your_slack_bot_token
SLACK_CHANNEL_ID=your_slack_channel_id

# Other services
BILLING_API_KEY=your_billing_api_key
VERIFICATION_API_KEY=your_verification_api_key
    

# requirements.txt (following your project dependencies)
xpander-sdk
agno
xpander-utils
python-dotenv
asyncio
python-dateutil
slack-sdk
certifi
requests
anthropic
    

customer_service_agent/
├── core/                   # Core business logic
│   ├── __init__.py
│   └── customer_service.py
├── tools/                  # AI tool wrappers
│   ├── __init__.py
│   └── customer_tools.py
├── orchestrator/           # Agent orchestration logic
│   ├── __init__.py
│   └── agno_agent.py
├── main.py                 # CLI entry point
├── xpander_handler.py      # Xpander event processing
├── xpander_config.json     # Xpander configuration
├── customer_support.db     # SQLite database
├── .env                    # Environment variables
├── requirements.txt        # Dependencies
└── Dockerfile              # Container setup (optional)
    

# Configure multiple models (following your project patterns)
def setup_multi_model_agent():
    """Setup agent with multiple model options"""
    
    # Primary model: Sonnet 4 for complex reasoning
    primary_model = Claude(
        id="claude-3-5-sonnet-20241022",
        api_key=os.getenv("ANTHROPIC_API_KEY")
    )
    
    # Quick response model: GPT for simple queries
    quick_model = OpenAI(
        id="gpt-4o-mini",
        api_key=os.getenv("OPENAI_API_KEY")
    )
    
    return {
        "primary": primary_model,
        "fallback": fallback_model,
        "quick": quick_model
    }
    

# Database management (following your pr_review.py pattern)
def manage_customer_data():
    """Database maintenance and optimization"""
    
    conn = sqlite3.connect("customer_support.db")
    cursor = conn.cursor()
    
    # Clean old interactions (retention policy)
    cutoff_date = (datetime.now() - timedelta(days=90)).isoformat()
    cursor.execute('''
        DELETE FROM customer_interactions 
        WHERE created_at < ? AND resolved = 1
    ''', (cutoff_date,))
    
    # Optimize database
    cursor.execute('VACUUM')
    
    # Generate analytics
    cursor.execute('''
        SELECT 
            COUNT(*) as total_tickets,
            AVG(CASE WHEN resolved = 1 THEN 1 ELSE 0 END) as resolution_rate,
            COUNT(DISTINCT customer_id) as unique_customers
        FROM customer_interactions
        WHERE created_at > ?
    ''', (cutoff_date,))
    
    stats = cursor.fetchone()
    conn.close()
    
    return {
        "total_tickets": stats[0],
        "resolution_rate": round(stats[1] * 100, 2),
        "unique_customers": stats[2]
    }
    

# Dockerfile (following your working Dockerfile)
FROM python:3.12-alpine AS builder

# Set environment vars for venv
ENV VIRTUAL_ENV=/venv
ENV PATH="$VIRTUAL_ENV/bin:$PATH"
ENV PYTHONUNBUFFERED=1

# Set working directory
WORKDIR /usr/src/app

# Install dependencies
RUN apk add --no-cache \
    curl \
    bash \
    gcc \
    g++ \
    libffi-dev \
    musl-dev \
    openssl-dev \
    make \
    python3-dev \
    git

# Create and activate virtualenv
RUN python3 -m venv $VIRTUAL_ENV

# Copy app files
COPY . .

# Install Python dependencies
RUN pip install --no-cache-dir --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

# Run the xpander handler
CMD ["python", "xpander_handler.py"]