---
title: "The 80% of agents no one shows you"
newsletter: "MLOps Community"
date: 2025-07-29
source: https://aaif.live/newsletters/mlopscommunity/2025-07-29-the-80-of-agents-no-one-shows-you
---

# The 80% of agents no one shows you

*Behind every LLM agent is a mess of infrastructure. Here’s one way teams are tackling it.*

*MLOps Community — Agentic AI Foundation, 2025-07-29*

Following Astonomer's hiring of Gwyneth Paltrow, former CEO Andy Byron has hired his own temporary spokesperson [https://go.mlops.community/daum6o].

## What You'll Learn Today:

XPANDER.AI: ONE WAY TO TACKLE THE AGENT INFRASTRUCTURE PROBLEM, BY ROHIT GHUMARE [https://www.linkedin.com/in/rohit-ghumare/]

What You'll Learn Today:

 * The Hidden Infrastructure Problem - Why 80% of AI agent work isn't the agent itself
 * xpander.ai Solution Overview - Backend-as-a-Service for production AI agents
 * Step-by-Step Implementation - From agent architecture to database setup
 * Complete Customer Service Agent - Working code with tools, memory, and event handling
 * Advanced Production Strategies - Multi-model support, database management, scaling
 * Framework Integration - Agno, Xpander SDK, and multiple AI model support
 * Production Deployment - CLI interface, event processing, and monitoring
 * Key Takeaways & Next Steps - Engineering insights and framework comparisons

Estimated Reading Time: 15-18 minutes
Implementation Time: 2-4 hours for basic setup, 1-2 days for production deployment

Three months ago, a startup with a team of five engineers was struggling to scale their AI customer service agent. The agent worked perfectly for single conversations, but when they tried to handle multiple users simultaneously, everything fell apart. Memory leaked between conversations, tools crashed under load, and state management became a nightmare.

Sound familiar? If you've ever tried to build production AI agents, you've probably hit the same wall.

The problem isn't your code, it's that building agent infrastructure is incredibly complex. You need memory management, tool orchestration, state persistence, multi-user isolation, event streaming, and guardrails. Most teams end up building all of this from scratch, which takes months and often breaks in production.

The Infrastructure Problem Nobody Talks About

Here's what most AI agent tutorials don't tell you: the agent itself is only 20% of the work. The other 80% is infrastructure:

 * Memory Management: How do you persist conversation history across sessions?
 * Tool Integration: How do you safely execute external API calls and code?
 * Multi-User State: How do you isolate user data and prevent cross-contamination?
 * Event Streaming: How do you handle real-time updates and notifications?
 * Guardrails: How do you prevent agents from doing dangerous things?
 * Scaling: How do you handle thousands of concurrent conversations?

Most teams spend 6-12 months building this infrastructure before they can focus on their actual AI product.

One Way to Solve It: xpander.ai as a Backend Layer

There are a few emerging approaches to this problem. Some teams rely on general-purpose orchestration frameworks like LangChain or CrewAI. Others use cloud-native infrastructure patterns and assemble their own stack.

xpander.ai is one managed approach: it provides backend infrastructure for AI agents as a service. If you're looking for something pre-built, here's what it offers:

 * Framework-agnostic integration (OpenAI, Anthropic, LangChain, CrewAI, etc.)
 * Persistent memory and conversation state
 * Tooling support with MCP compatibility
 * Multi-user state management with isolation
 * Real-time event streaming
 * Built-in guardrails and security
 * Agent-to-agent communication

This kind of setup can significantly cut down the time it takes to get a production-ready agent live - sometimes reducing months of work to just a few days, depending on your architecture and use case.

Step-by-Step: Setting Up xpander.ai for Production Agents

Let's build a customer service agent with xpander.ai and agno framework.

1. Define Your Agent Architecture

Before diving into code, plan your agent's capabilities:

Core Agent Functions

 * Conversation handling and memory persistence
 * Tool access and external API integration
 * User state management and personalization
 * Event handling and real-time updates

Infrastructure Requirements

 * Multi-user isolation and data security
 * Scalable hosting and load balancing
 * Monitoring and observability
 * Backup and disaster recovery

2. Install and Configure xpander SDK

Install the xpander SDK for your preferred language:

# 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

Initialize your xpander client following the actual project patterns:

Python # 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)

3. Implement Memory and State Management

Set up persistent memory using real Agno framework patterns:

# 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()

4. Integrate Tools and External APIs

Connect your agent to external services using real Agno tool patterns:

# 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())

5. Set Up Main Execution and Event Handling

Create the main execution files following your working project patterns:

# 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")

6. Configuration and Environment Setup

Create the required configuration files:

// 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

7. Project Structure

Organize your files following the working project structure:

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)

Advanced Production Strategies

Multiple Model Support

Configure your agent to use different models based on requirements:

# 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

Implement proper database management following your project patterns:

# 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] }

Docker Deployment

Create a Dockerfile following your project pattern:

# 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"]


PRODUCTION DEPLOYMENT CHECKLIST

Pre-Production Setup

 * Configure environment variables in .env
 * Set up xpander_config.json with real credentials
 * Test all tools with python main.py
 * Initialize database with proper permissions
 * Configure Slack integration (optional)

Security Configuration

 * Enable database encryption
 * Set up proper API key rotation
 * Configure rate limiting for tools
 * Implement input validation and sanitization
 * Set up audit logging

Monitoring and Observability

 * Configure error tracking and alerting
 * Set up performance monitoring
 * Implement customer satisfaction tracking
 * Monitor token usage and costs
 * Set up backup and recovery procedures

Scaling Considerations

 * Configure auto-scaling policies
 * Set up load balancing if needed
 * Implement caching strategies
 * Monitor database performance
 * Plan for traffic spikes

Follow the progress of the xpander.ai GitHub Repository [https://github.com/xpander-ai/xpander.ai] to stay updated, and if you find it valuable, consider giving it a star ⭐️.

Framework Integrations:

 * OpenAI SDK Integration [https://platform.openai.com/docs/api-reference/introduction]: Compatible with OpenAI's latest APIs
 * LangChain Support [https://www.langchain.com/]: Works with LangChain agent frameworks
 * CrewAI Integration [https://www.crewai.com/]: Supports CrewAI multi-agent workflows
 * Anthropic Claude [https://claude.ai/]: Compatible with Claude API
 * Google AI (Gemini) [https://gemini.google.com/]: Works with Google's AI models
 * Motia: Unified Backend Framework [https://github.com/MotiaDev/motia?tab=readme-ov-file]: Modern backend for APIs, events, and AI agents

Production-Ready Features:

 * Database Management: SQLite with proper schema design
 * Error Handling: Comprehensive try/catch blocks
 * Event Processing: Real Xpander event integration
 * Tool Integration: Working @tool patterns with external APIs
 * Memory Management: Persistent conversation history
 * Multi-Model Support: Flexible model selection

Development Workflow:

 * Start with CLI: Use python main.py for development
 * Test Tools: Verify each @tool function works correctly
 * Deploy with Events: Use python xpander_handler.py for production
 * Monitor Performance: Track metrics and user satisfaction
 * Scale Gradually: Add more tools and models as needed


KEY TAKEAWAYS

For Engineering Teams:

 * Infrastructure Dominates: Most of the work in AI agents lies in infrastructure - memory, orchestration, persistence, scaling - not just model prompts.
 * Faster Setup: Tools like xpander.ai can significantly reduce build time, especially for teams starting from scratch.
 * Production Considerations: Includes components like scaling, monitoring, and state isolation that are often missing in early-stage builds.
 * Framework Compatibility: Can integrate with OpenAI, Anthropic, LangChain, CrewAI, or custom implementations.

For Technical Leaders:

 * Cost Awareness: Model routing and caching strategies may help reduce operational costs depending on usage patterns.
 * Security Context: Guardrails, audit logging, and isolation features can support teams working in regulated or multi-user environments.
 * Focus and Velocity: Moving common infrastructure concerns into reusable components can free engineers to focus on business logic.
 * Scalability Options: Built-in autoscaling can help handle bursty workloads without additional orchestration.

Framework Integration:

 * Agno Framework: Agent orchestration and tool interface layer
 * Xpander SDK: Handles backend communication, event routing, and configuration
 * Model Flexibility: Tested with Claude Sonnet 4, GPT-4o, and others
 * Database Support: Uses SQLite with structured schema for persistent state

Next Steps:

 * Try running the examples in the xpander.ai repository to test local setup
 * Explore how your existing monitoring stack could plug into an agent deployment
 * Contribute issues, improvements, or feedback to the open-source project as you go

Agent infrastructure isn’t solved yet, but reusable infrastructure can take you a long way.

Working on something tricky or planning ahead? Here’s how we can help - just hit reply:

 * Custom workshops tailored to your company’s needs
 * Hiring? I know some quality folks looking for a new adventure
 * Want to connect with someone tackling similar problems? I can introduce you

Thanks for reading, catch you next time!

Interested in partnering with us? Get in touch: partners@mlops.community

Thanks for reading. See you in Slack [https://go.mlops.community/slack], YouTube [https://www.youtube.com/channel/UCG6qpjVnBTTT8wLGBygANOQ?view_as=subscriber], and podcast [https://home.mlops.community/public/content/] land. Oh yeah, and we are also on X [https://twitter.com/mlopscommunity] and LinkedIn [https://go.mlops.community/linkedin].

Thanks again to Rohit Ghumare [https://www.linkedin.com/in/rohit-ghumare/] for his contribution.

---
Source: https://aaif.live/newsletters/mlopscommunity/2025-07-29-the-80-of-agents-no-one-shows-you