Skip to main content

Scaling and Architecture

Calculating read time…

You've built your first MCP server. It works beautifully with one AI model. Then your company decides to integrate five different AI models—GPT-4, Claude, Gemini, local Llama, and a custom fine-tuned model.

Each model needs different tools. Some share tools but with different permissions. The context needs vary wildly. Your single-server approach starts to crack under the complexity.

Welcome to enterprise-scale MCP.

This isn't about making tools work anymore—it's about architecting systems that scale across multiple models, teams, and use cases while remaining maintainable, secure, and performant.

In this guide, you'll master the advanced concepts that separate hobbyist MCP implementations from production-grade, enterprise-ready architectures: context engineering, multi-model server design, tool discoverability patterns, and sophisticated naming conventions! 🏗️

🎯 The Scale Problem: When Simple Solutions Break

Before diving into solutions, let's understand what breaks when you scale MCP beyond toy examples.

The Five Scaling Challenges

❌ Challenge #1: Context Overload

You give the AI access to 50 tools.
The tool descriptions alone consume 10,000 tokens.
The AI gets confused, chooses wrong tools, or hits context limits.
Result: Poor performance and wasted API costs.
❌ Challenge #2: Model-Specific Requirements

GPT-4 expects function descriptions in one format.
Claude prefers different parameter structures.
Your local Llama model needs simpler tools.
Result: One-size-fits-all doesn't fit anyone well.
❌ Challenge #3: Tool Discovery Chaos

Tools have inconsistent names: getUserData, fetch_user, user-info
No clear categorization or grouping.
AI can't find the right tool for the task.
Result: Frustrated users and failed operations.
❌ Challenge #4: Monolithic Server Bloat

One giant server handles database, files, APIs, email, everything.
Changes require redeploying the entire system.
Different teams step on each other's toes.
Result: Unmaintainable mess and deployment paralysis.
❌ Challenge #5: Permission Complexity

Some models should read-only, others full access.
Different users need different tool subsets.
Audit logging across models is inconsistent.
Result: Security holes and compliance nightmares.

Let's solve each of these with expert-level patterns!

Context Engineering: Optimizing AI Performance

Context engineering is the art of giving AI models exactly the information they need—no more, no less.

The Context Budget Problem

Every AI model has a context window limit:

  • GPT-4: 128K tokens (~96,000 words)
  • Claude 3 Opus: 200K tokens (~150,000 words)
  • Local Llama 3: 8K tokens (~6,000 words)

But context isn't free! Each token costs money and affects performance:

Cost Breakdown (GPT-4 Example): ───────────────────────────────── Input: $10 per 1M tokens Output: $30 per 1M tokens Scenario: 50 tools × 200 tokens each = 10,000 tokens Cost per call: $0.10 input + $0.30 output 1000 calls/day: $400/day = $12,000/month Just on tool descriptions!

Strategy 1: Lazy Tool Loading

Don't send all tools upfront. Send tools only when needed. 

"""
Lazy Tool Loading Pattern
Tools are discovered and loaded on-demand
"""

from mcp.server.fastmcp import FastMCP, Context
from typing import List, Optional

class LazyToolRegistry:
    """Manages tool loading based on context"""
    
    def __init__(self):
        self.all_tools = {
            "database": ["query_users", "update_record", "get_stats"],
            "files": ["read_file", "write_file", "list_directory"],
            "email": ["send_email", "search_inbox", "get_thread"],
            "analytics": ["generate_report", "export_data", "run_query"]
        }
    
    def get_tools_for_query(self, query: str) -> List[str]:
        """
        Intelligently select relevant tools based on query
        """
        query_lower = query.lower()
        
        relevant_tools = []
        
        # Database keywords
        if any(word in query_lower for word in ["user", "record", "data", "query", "database"]):
            relevant_tools.extend(self.all_tools["database"])
        
        # File keywords  
        if any(word in query_lower for word in ["file", "read", "write", "directory", "folder"]):
            relevant_tools.extend(self.all_tools["files"])
        
        # Email keywords
        if any(word in query_lower for word in ["email", "message", "inbox", "send"]):
            relevant_tools.extend(self.all_tools["email"])
        
        # Analytics keywords
        if any(word in query_lower for word in ["report", "analytics", "export", "chart"]):
            relevant_tools.extend(self.all_tools["analytics"])
        
        # Default: if nothing matches, load common tools
        if not relevant_tools:
            relevant_tools = self.all_tools["database"][:2]  # Basic tools
        
        return relevant_tools

# Usage in MCP server
mcp = FastMCP("Smart Tool Server")
registry = LazyToolRegistry()

@mcp.tool()
async def discover_tools(query: str, ctx: Context) -> dict:
    """
    Dynamically discover relevant tools for a query
    
    The AI calls this first to find which tools it needs,
    then calls those specific tools.
    """
    relevant_tools = registry.get_tools_for_query(query)
    
    await ctx.info(f"Found {len(relevant_tools)} relevant tools")
    
    return {
        "query": query,
        "relevant_tools": relevant_tools,
        "total_available": sum(len(v) for v in registry.all_tools.values())
    }

Strategy 2: Context Compression

Compress tool descriptions without losing meaning.

❌ Verbose Description (200 tokens):

"This tool allows you to query the user database to retrieve information about users. You can search by various fields including user ID, email address, username, or registration date. The tool returns a comprehensive user object containing all profile information, preferences, activity history, and related metadata. Use this tool whenever you need to access user information from the database."
Compressed Description (40 tokens):

"Query users by ID/email/username. Returns: profile, preferences, history. Use: get user info."

Savings: 80% fewer tokens, same information!

Strategy 3: Hierarchical Tool Organization

Organize tools into categories with summary tools. 

"""
Hierarchical Tool Discovery
AI first discovers categories, then tools within categories
"""

@mcp.tool()
async def list_tool_categories(ctx: Context) -> dict:
    """
    Get available tool categories.
    Call this first to explore what's available.
    """
    return {
        "categories": [
            {
                "name": "database",
                "description": "Query and update user/product data",
                "tool_count": 5
            },
            {
                "name": "files",
                "description": "Read/write files and directories",
                "tool_count": 4
            },
            {
                "name": "communication",
                "description": "Email, Slack, notifications",
                "tool_count": 6
            }
        ]
    }

@mcp.tool()
async def list_category_tools(category: str, ctx: Context) -> dict:
    """
    Get tools in a specific category.
    Call after discovering categories.
    """
    category_tools = {
        "database": [
            {"name": "query_users", "desc": "Search users"},
            {"name": "update_record", "desc": "Modify records"},
            {"name": "get_stats", "desc": "Get statistics"}
        ],
        "files": [
            {"name": "read_file", "desc": "Read file contents"},
            {"name": "write_file", "desc": "Write to file"}
        ]
    }
    
    return {
        "category": category,
        "tools": category_tools.get(category, [])
    }
🎓 Context Engineering Workflow:

Step 1: AI gets query from user
Step 2: AI calls discover_tools(query) → Gets 5 relevant tools
Step 3: Only those 5 tools sent to AI's context
Step 4: AI uses appropriate tool

Result: 90% context reduction, faster responses, lower costs!

🔄 Multi-Model MCP Servers: Serving Different AI Models

Different AI models have different strengths, costs, and capabilities. Your MCP architecture should serve them all efficiently.

The Multi-Model Architecture Pattern

┌─────────────────────────────────────────────────┐ │ HOST APPLICATION │ │ (Your AI Platform / Orchestrator) │ └───────────┬─────────────────────────────────────┘ │ ├─────────────┬─────────────┬─────────────┐ │ │ │ │ ⬍ ⬍ ⬍ ⬍ ┌───────────▼──┐ ┌───────▼──┐ ┌───────▼──┐ ┌───────▼──┐ │ GPT-4 Client│ │Claude Cl │ │Gemini Cl │ │Llama Cl │ │ │ │ient │ │ient │ │ient │ └───────┬──────┘ └─────┬────┘ └─────┬────┘ └─────┬────┘ │ │ │ │ ├────────────────┴──────────────┴──────────────┤ │ │ ⬍ Each connects to appropriate MCP servers ⬍ │ │ ┌───────▼────────┬─────────────┬──────────────┬────────▼───┐ │ │ │ │ │ │ Production │ Analytics │ Read-Only │ Simple │ │ Tools Server │ Server │ Server │ Tools │ │ (GPT-4, │ (Claude) │ (All) │ (Llama) │ │ Claude) │ │ │ │ └────────────────┴─────────────┴──────────────┴────────────┘

Pattern 1: Model-Specific Tool Exposure

Different models see different subsets of tools based on their capabilities. 

"""
Model-Aware MCP Server
Exposes different tools to different models
"""

from mcp.server.fastmcp import FastMCP, Context
from enum import Enum

class ModelType(Enum):
    GPT4 = "gpt-4"
    CLAUDE = "claude-3-opus"
    GEMINI = "gemini-pro"
    LLAMA = "llama-3-70b"

class ModelAwareMCPServer:
    """Server that adapts to different AI models"""
    
    def __init__(self):
        # Define tool access by model capability
        self.model_tools = {
            ModelType.GPT4: {
                "tools": ["complex_analysis", "multi_step_workflow", 
                         "advanced_calculation", "database_query"],
                "max_complexity": "high",
                "parallel_calls": True
            },
            ModelType.CLAUDE: {
                "tools": ["document_analysis", "code_review",
                         "reasoning_task", "database_query"],
                "max_complexity": "high",
                "parallel_calls": True
            },
            ModelType.GEMINI: {
                "tools": ["image_analysis", "search_query",
                         "simple_calculation", "database_query"],
                "max_complexity": "medium",
                "parallel_calls": True
            },
            ModelType.LLAMA: {
                "tools": ["simple_calculation", "text_processing"],
                "max_complexity": "low",
                "parallel_calls": False
            }
        }
    
    def get_tools_for_model(self, model: ModelType) -> dict:
        """Get available tools for specific model"""
        config = self.model_tools.get(model, self.model_tools[ModelType.LLAMA])
        return config

# Create model-specific servers
mcp_gpt4 = FastMCP("MCP for GPT-4")
mcp_claude = FastMCP("MCP for Claude")
mcp_llama = FastMCP("MCP for Llama")

# Complex tool - only for capable models
@mcp_gpt4.tool()
@mcp_claude.tool()
async def complex_analysis(
    data: dict,
    analysis_type: str,
    ctx: Context
) -> dict:
    """
    Perform complex multi-step analysis.
    Requires: High reasoning capability
    """
    await ctx.info("Running complex analysis...")
    # Complex logic here
    return {"result": "analysis complete"}

# Simple tool - available to all models
@mcp_gpt4.tool()
@mcp_claude.tool()
@mcp_llama.tool()
async def simple_calculation(a: float, b: float, op: str) -> dict:
    """
    Basic arithmetic: add, subtract, multiply, divide
    Works with all models
    """
    ops = {
        "add": a + b,
        "subtract": a - b,
        "multiply": a * b,
        "divide": a / b if b != 0 else None
    }
    return {"result": ops.get(op)}

Pattern 2: Adaptive Tool Descriptions

Same tool, but description complexity varies by model. 

"""
Adaptive Tool Descriptions
Simple descriptions for simple models,
detailed ones for advanced models
"""

class AdaptiveToolDescriptor:
    """Generates model-appropriate tool descriptions"""
    
    @staticmethod
    def get_description(tool_name: str, model: ModelType) -> str:
        """
        Return description complexity based on model capability
        """
        
        descriptions = {
            "database_query": {
                ModelType.GPT4: """
                    Execute SQL queries on production database.
                    Supports: SELECT, JOIN, aggregations, subqueries.
                    Parameters:
                    - query: SQL string (validated for safety)
                    - limit: Max rows (default 100, max 1000)
                    - timeout: Query timeout in seconds
                    Returns: rows as list of dicts, metadata
                """,
                ModelType.LLAMA: """
                    Run database query.
                    Args: query (SQL), limit (max rows)
                    Returns: data rows
                """
            }
        }
        
        tool_descs = descriptions.get(tool_name, {})
        return tool_descs.get(model, tool_descs.get(ModelType.LLAMA))

# Usage
descriptor = AdaptiveToolDescriptor()

# For GPT-4: Detailed description
gpt4_desc = descriptor.get_description("database_query", ModelType.GPT4)
# → Long, detailed, with examples

# For Llama: Simple description  
llama_desc = descriptor.get_description("database_query", ModelType.LLAMA)
# → Short, essential info only

Pattern 3: Model Orchestration Layer

Route requests to the right model based on task complexity. 

"""
Multi-Model Orchestrator
Automatically selects best model for each task
"""

class ModelOrchestrator:
    """Routes tasks to optimal models"""
    
    def __init__(self):
        self.model_costs = {
            ModelType.GPT4: 0.03,      # $ per 1K tokens
            ModelType.CLAUDE: 0.015,
            ModelType.GEMINI: 0.001,
            ModelType.LLAMA: 0.0        # Free (local)
        }
        
        self.model_capabilities = {
            "complex_reasoning": [ModelType.GPT4, ModelType.CLAUDE],
            "code_analysis": [ModelType.CLAUDE, ModelType.GPT4],
            "simple_tasks": [ModelType.LLAMA, ModelType.GEMINI],
            "image_tasks": [ModelType.GEMINI, ModelType.GPT4]
        }
    
    def select_model(
        self,
        task_type: str,
        budget: float = None,
        priority: str = "balanced"
    ) -> ModelType:
        """
        Select best model for task
        
        Args:
            task_type: Type of task
            budget: Max cost willing to spend
            priority: "cost", "quality", or "balanced"
        """
        
        # Get capable models for this task
        capable_models = self.model_capabilities.get(
            task_type,
            [ModelType.LLAMA]
        )
        
        if priority == "cost":
            # Choose cheapest capable model
            return min(capable_models, key=lambda m: self.model_costs[m])
        
        elif priority == "quality":
            # Choose most capable (first in list)
            return capable_models[0]
        
        else:  # balanced
            # Mid-tier option
            return capable_models[len(capable_models) // 2]

# Usage example
orchestrator = ModelOrchestrator()

# Simple task → Use cheap model
model = orchestrator.select_model("simple_tasks", priority="cost")
# → Returns: ModelType.LLAMA (free!)

# Complex task → Use best model
model = orchestrator.select_model("complex_reasoning", priority="quality")
# → Returns: ModelType.GPT4 (most capable)
✅ Multi-Model Benefits:

• Cost optimization: Use expensive models only when needed
• Performance: Route to fastest model for simple tasks
• Reliability: Fallback to alternate models if primary fails
• Flexibility: Easy to add new models to the system

🔍 Tool Discoverability & Naming Conventions

Good naming is the difference between AI finding the right tool instantly vs. fumbling through dozens of options.

The Naming Convention Standard

A well-named tool tells AI exactly what it does without reading the description.

Tool Naming Pattern:

[namespace]_[resource]_[action]_[modifier?]

Examples:
db_users_query_active → Database, users table, query active ones
api_weather_fetch_current → API, weather data, fetch current
file_documents_read_recent → Files, documents folder, read recent
email_inbox_search_unread → Email, inbox, search unread

Pattern 1: Hierarchical Namespacing

❌ Bad Naming (Flat, Ambiguous):

getUser → Get from where? Database? API? Cache?
sendMessage → Via email? Slack? SMS?
search → Search what? Where?
calculate → Calculate what?
✅ Good Naming (Hierarchical, Clear):

database_users_get_by_id → Crystal clear
communication_slack_send_message → Obvious
search_documents_by_keywords → Specific
math_statistics_calculate_mean → No confusion

Pattern 2: Verb Consistency

Use consistent verbs across all tools:

Operation Standard Verb Example
Read data get, read, fetch db_users_get
Create data create, add, insert db_users_create
Modify data update, modify, edit db_users_update
Remove data delete, remove db_users_delete
Search/filter search, find, query db_users_search
List all list, get_all db_users_list

Pattern 3: Searchable Tool Metadata

Add rich metadata that helps AI discover tools semantically. 

"""
Tool with Rich Metadata for Discovery
"""

from mcp.server.fastmcp import FastMCP
from typing import List

mcp = FastMCP("Discoverable Tools Server")

@mcp.tool()
async def database_users_search_by_email(
    email: str,
    include_inactive: bool = False
) -> dict:
    """
    Search for user accounts by email address.
    
    Category: Database Operations > User Management
    Tags: user, email, search, lookup, account
    Use When: Need to find user by their email address
    Examples:
    - "Find the user with email john@example.com"
    - "Look up account for sarah@company.org"
    - "Get user details for email address"
    
    Args:
        email: Email address to search for
        include_inactive: Whether to include deactivated accounts
    
    Returns:
        User object with profile, preferences, and activity
    """
    # Implementation
    pass

# Tool registry with semantic search
class SemanticToolRegistry:
    """Enables semantic tool discovery"""
    
    def __init__(self):
        self.tool_index = {
            "database_users_search_by_email": {
                "name": "database_users_search_by_email",
                "category": "database/user_management",
                "tags": ["user", "email", "search", "lookup", "account"],
                "synonyms": ["find user", "lookup account", "email search"],
                "related_tools": [
                    "database_users_get_by_id",
                    "database_users_search_by_name"
                ]
            }
        }
    
    def find_tools(self, query: str) -> List[str]:
        """Find tools matching query keywords"""
        query_lower = query.lower()
        matches = []
        
        for tool_name, metadata in self.tool_index.items():
            # Check tags
            if any(tag in query_lower for tag in metadata["tags"]):
                matches.append(tool_name)
            # Check synonyms
            elif any(syn in query_lower for syn in metadata["synonyms"]):
                matches.append(tool_name)
        
        return matches

# Usage
registry = SemanticToolRegistry()

# AI asks: "How do I find a user by their email?"
tools = registry.find_tools("find user email")
# → Returns: ["database_users_search_by_email"]
💡 Naming Best Practices:

• Use underscores, not camelCase: get_user_data not getUserData
• Keep names under 50 characters
• Start with namespace (domain area)
• End with action verb
• No abbreviations unless universally known (API, DB, ID okay)
• Include resource type in name
• Add modifiers for specificity when needed

🏗️ Exercise: Design Multi-Model Server Architecture

Time to apply everything! Let's design a complete MCP architecture for a real company.

Scenario: E-Commerce Platform

You're the lead architect for ShopAI, an e-commerce company integrating AI across their platform.

Requirements:

  • Models in use: GPT-4 (customer service), Claude (product analysis), Llama (simple tasks)
  • Data sources: PostgreSQL (products, orders, users), MongoDB (reviews), Redis (cache)
  • External APIs: Stripe (payments), SendGrid (email), Twilio (SMS)
  • Tools needed: 30+ different operations
  • Teams: Customer support, analytics, engineering, marketing

Solution: Layered Multi-Model Architecture

┌──────────────────────────────────────────────────────────┐ │ AI ORCHESTRATION LAYER │ │ ┌─────────────┬──────────────┬──────────────┐ │ │ │ GPT-4 │ Claude │ Llama │ │ │ │ (Customer │ (Analytics) │ (Simple) │ │ │ │ Service) │ │ │ │ │ └──────┬──────┴──────┬───────┴──────┬───────┘ │ └─────────┼─────────────┼──────────────┼──────────────────┘ │ │ │ ⬍ ⬍ ⬍ ┌─────────▼─────────────▼──────────────▼──────────────────┐ │ MCP GATEWAY / ROUTER │ │ • Auth & permissions │ │ • Request routing │ │ • Load balancing │ │ • Caching │ └─────┬──────────┬──────────┬──────────┬──────────────────┘ │ │ │ │ ⬍ ⬍ ⬍ ⬍ ┌─────▼────┬─────▼────┬─────▼────┬─────▼──────┐ │ Customer │ Product │ Order │ Payment │ │ Server │ Server │ Server │ Server │ ├──────────┼──────────┼──────────┼────────────┤ │ - search │ - search │ - create │ - charge │ │ - get │ - get │ - update │ - refund │ │ - update │ - analyze│ - cancel │ - status │ │ - create │ - review │ - history│ │ └──────────┴──────────┴──────────┴────────────┘ │ │ │ │ ⬍ ⬍ ⬍ ⬍ ┌─────▼──────────▼──────────▼──────────▼────────┐ │ DATA & SERVICE LAYER │ │ PostgreSQL MongoDB Redis Stripe SendGrid │ └────────────────────────────────────────────────┘

Implementation: Server Layout

Create shopai_architecture.py: 

"""
ShopAI Multi-Model MCP Architecture
Production-ready e-commerce AI integration
"""

from mcp.server.fastmcp import FastMCP, Context
from typing import List, Optional
from enum import Enum

# ============================================
# CONFIGURATION
# ============================================

class ServerDomain(Enum):
    """Server organization by domain"""
    CUSTOMER = "customer"
    PRODUCT = "product"
    ORDER = "order"
    PAYMENT = "payment"
    ANALYTICS = "analytics"

class ModelAccess(Enum):
    """Model access levels"""
    READ_ONLY = "read"
    READ_WRITE = "write"
    ADMIN = "admin"

# Model-to-server access matrix
MODEL_PERMISSIONS = {
    "gpt-4": {
        ServerDomain.CUSTOMER: ModelAccess.READ_WRITE,
        ServerDomain.PRODUCT: ModelAccess.READ_ONLY,
        ServerDomain.ORDER: ModelAccess.READ_WRITE,
        ServerDomain.PAYMENT: ModelAccess.READ_ONLY,
    },
    "claude-3-opus": {
        ServerDomain.PRODUCT: ModelAccess.READ_WRITE,
        ServerDomain.ANALYTICS: ModelAccess.READ_WRITE,
        ServerDomain.CUSTOMER: ModelAccess.READ_ONLY,
    },
    "llama-3-70b": {
        ServerDomain.PRODUCT: ModelAccess.READ_ONLY,
        ServerDomain.CUSTOMER: ModelAccess.READ_ONLY,
    }
}

# ============================================
# CUSTOMER SERVER (Used by GPT-4)
# ============================================

customer_server = FastMCP("ShopAI Customer Server")

@customer_server.tool()
async def customer_users_search_by_email(
    email: str,
    ctx: Context
) -> dict:
    """
    Search customer by email address.
    
    Category: Customer Management > Search
    Access: Read-Write (GPT-4), Read-Only (Llama)
    """
    await ctx.info(f"Searching for customer: {email}")
    
    # Database query
    customer = {
        "id": "cust_123",
        "email": email,
        "name": "John Doe",
        "orders": 15,
        "lifetime_value": 1250.00
    }
    
    return {
        "success": True,
        "customer": customer
    }

@customer_server.tool()
async def customer_support_create_ticket(
    customer_id: str,
    issue: str,
    priority: str,
    ctx: Context
) -> dict:
    """
    Create customer support ticket.
    
    Category: Customer Management > Support
    Access: Read-Write (GPT-4 only)
    """
    await ctx.info("Creating support ticket...")
    
    ticket = {
        "ticket_id": "TKT-12345",
        "customer_id": customer_id,
        "issue": issue,
        "priority": priority,
        "status": "open",
        "assigned_to": "auto_assign"
    }
    
    return {
        "success": True,
        "ticket": ticket
    }

# ============================================
# PRODUCT SERVER (Used by Claude, All models read)
# ============================================

product_server = FastMCP("ShopAI Product Server")

@product_server.tool()
async def product_catalog_search_by_keywords(
    keywords: List[str],
    category: Optional[str] = None,
    ctx: Context
) -> dict:
    """
    Search product catalog by keywords.
    
    Category: Product Management > Search
    Access: All models (Read-Only)
    """
    await ctx.info(f"Searching products: {keywords}")
    
    products = [
        {
            "id": "prod_001",
            "name": "Wireless Headphones",
            "price": 79.99,
            "stock": 150,
            "rating": 4.5
        },
        {
            "id": "prod_002",
            "name": "Bluetooth Speaker",
            "price": 49.99,
            "stock": 89,
            "rating": 4.2
        }
    ]
    
    return {
        "success": True,
        "products": products,
        "total": len(products)
    }

@product_server.tool()
async def product_analytics_analyze_trends(
    time_period: str,
    category: str,
    ctx: Context
) -> dict:
    """
    Analyze product sales trends and patterns.
    
    Category: Product Management > Analytics
    Access: Read-Write (Claude), Read-Only (GPT-4)
    Complexity: High (not available to Llama)
    """
    await ctx.info(f"Analyzing trends for {category}...")
    
    analysis = {
        "time_period": time_period,
        "category": category,
        "top_products": ["prod_001", "prod_005", "prod_012"],
        "growth_rate": 12.5,
        "seasonal_patterns": {
            "peak_months": ["November", "December"],
            "low_months": ["February", "March"]
        },
        "recommendations": [
            "Increase inventory for top products",
            "Run promotions in low months"
        ]
    }
    
    return {
        "success": True,
        "analysis": analysis
    }

# ============================================
# ORDER SERVER (Used by GPT-4)
# ============================================

order_server = FastMCP("ShopAI Order Server")

@order_server.tool()
async def order_management_create_order(
    customer_id: str,
    items: List[dict],
    ctx: Context
) -> dict:
    """
    Create new customer order.
    
    Category: Order Management > Create
    Access: Read-Write (GPT-4 only)
    """
    await ctx.info("Creating order...")
    
    total = sum(item["price"] * item["quantity"] for item in items)
    
    order = {
        "order_id": "ORD-78901",
        "customer_id": customer_id,
        "items": items,
        "total": total,
        "status": "pending",
        "created_at": "2024-02-15T10:30:00Z"
    }
    
    return {
        "success": True,
        "order": order
    }

@order_server.tool()
async def order_management_get_history(
    customer_id: str,
    limit: int = 10,
    ctx: Context
) -> dict:
    """
    Get customer order history.
    
    Category: Order Management > Retrieve
    Access: Read-Only (All models)
    """
    await ctx.info(f"Fetching order history for {customer_id}")
    
    orders = [
        {
            "order_id": "ORD-78900",
            "date": "2024-02-10",
            "total": 129.99,
            "status": "delivered"
        },
        {
            "order_id": "ORD-78850",
            "date": "2024-01-28",
            "total": 45.50,
            "status": "delivered"
        }
    ]
    
    return {
        "success": True,
        "customer_id": customer_id,
        "orders": orders[:limit],
        "total_orders": len(orders)
    }

# ============================================
# PAYMENT SERVER (Read-only for AI safety)
# ============================================

payment_server = FastMCP("ShopAI Payment Server")

@payment_server.tool()
async def payment_transactions_get_status(
    transaction_id: str,
    ctx: Context
) -> dict:
    """
    Get payment transaction status.
    
    Category: Payment > Status Check
    Access: Read-Only (GPT-4, Claude)
    Security: High - no modification allowed
    """
    await ctx.info(f"Checking transaction: {transaction_id}")
    
    transaction = {
        "transaction_id": transaction_id,
        "amount": 79.99,
        "status": "completed",
        "payment_method": "card_****_4242",
        "timestamp": "2024-02-15T10:25:00Z"
    }
    
    return {
        "success": True,
        "transaction": transaction
    }

# Note: No create/update tools for payments
# Critical operations require human approval

# ============================================
# TOOL DISCOVERY SERVER
# ============================================

discovery_server = FastMCP("ShopAI Tool Discovery")

@discovery_server.tool()
async def discovery_list_servers(ctx: Context) -> dict:
    """
    List all available MCP servers and their domains.
    AI should call this first to understand the architecture.
    """
    servers = [
        {
            "name": "customer_server",
            "domain": "customer",
            "description": "Customer data and support",
            "tool_count": 5
        },
        {
            "name": "product_server",
            "domain": "product",
            "description": "Product catalog and analytics",
            "tool_count": 8
        },
        {
            "name": "order_server",
            "domain": "order",
            "description": "Order management",
            "tool_count": 6
        },
        {
            "name": "payment_server",
            "domain": "payment",
            "description": "Payment status (read-only)",
            "tool_count": 2
        }
    ]
    
    return {
        "servers": servers,
        "total": len(servers)
    }

@discovery_server.tool()
async def discovery_get_tools_for_domain(
    domain: str,
    ctx: Context
) -> dict:
    """
    Get all tools available in a specific domain.
    
    Args:
        domain: customer, product, order, or payment
    """
    domain_tools = {
        "customer": [
            "customer_users_search_by_email",
            "customer_users_get_by_id",
            "customer_support_create_ticket"
        ],
        "product": [
            "product_catalog_search_by_keywords",
            "product_catalog_get_by_id",
            "product_analytics_analyze_trends"
        ],
        "order": [
            "order_management_create_order",
            "order_management_get_history",
            "order_management_update_status"
        ],
        "payment": [
            "payment_transactions_get_status"
        ]
    }
    
    return {
        "domain": domain,
        "tools": domain_tools.get(domain, [])
    }

# ============================================
# RUNNING THE SERVERS
# ============================================

if __name__ == "__main__":
    import sys
    
    # Determine which server to run
    if len(sys.argv) < 2:
        print("Usage: python shopai_architecture.py [customer|product|order|payment|discovery]")
        sys.exit(1)
    
    server_type = sys.argv[1].lower()
    
    servers = {
        "customer": customer_server,
        "product": product_server,
        "order": order_server,
        "payment": payment_server,
        "discovery": discovery_server
    }
    
    server = servers.get(server_type)
    if server:
        print(f"🚀 Starting {server_type} server...", file=sys.stderr)
        server.run()
    else:
        print(f"❌ Unknown server: {server_type}", file=sys.stderr)
        sys.exit(1)

Deployment Configuration

Create shopai_deployment.yaml: 

# ShopAI MCP Deployment Configuration

services:
  # Discovery server - always available
  discovery:
    server: discovery_server
    port: 3000
    replicas: 2
    models: [gpt-4, claude, llama]
    
  # Customer server - GPT-4 primary
  customer:
    server: customer_server
    port: 3001
    replicas: 3
    models: [gpt-4, llama]
    rate_limit: 100/minute
    
  # Product server - Claude primary
  product:
    server: product_server
    port: 3002
    replicas: 3
    models: [claude, gpt-4, llama]
    rate_limit: 200/minute
    
  # Order server - GPT-4 only
  order:
    server: order_server
    port: 3003
    replicas: 2
    models: [gpt-4]
    rate_limit: 50/minute
    requires_approval: true
    
  # Payment server - Read-only
  payment:
    server: payment_server
    port: 3004
    replicas: 2
    models: [gpt-4, claude]
    rate_limit: 20/minute
    read_only: true

model_routing:
  gpt-4:
    primary_domains: [customer, order]
    allowed_domains: [product, payment]
    
  claude:
    primary_domains: [product, analytics]
    allowed_domains: [customer]
    
  llama:
    primary_domains: []
    allowed_domains: [customer, product]
    complexity_limit: low
✅ Architecture Benefits:

• Separation of concerns: Each server has one clear domain
• Model optimization: Right model for right task
• Security: Payment server is read-only, critical ops need approval
• Scalability: Scale servers independently based on load
• Maintainability: Teams own their domains
• Discovery: AI can explore available capabilities

📊 Expert Best Practices Summary

Context Engineering:

✓ Use lazy loading - send only needed tools
✓ Compress descriptions - 80% shorter, same info
✓ Hierarchical discovery - categories then tools
✓ Track context usage - monitor token consumption
Multi-Model Servers:

✓ Model-specific tool exposure
✓ Adaptive descriptions by capability
✓ Orchestration layer for routing
✓ Cost-based model selection
Tool Discoverability:

✓ Hierarchical naming: namespace_resource_action
✓ Consistent verbs across all tools
✓ Rich metadata with tags and examples
✓ Semantic search capabilities
Architecture Patterns:

✓ Domain-driven server organization
✓ Gateway/router for centralized control
✓ Model-to-server permission matrix
✓ Discovery server for exploration

📚 Essential Resources

Architecture Patterns:
• MCP Best Practices: modelcontextprotocol.info/docs/best-practices
• Multi-Agent Systems: Research papers on agent coordination
• Gateway Patterns: ContextForge (IBM open-source)

Tools & Libraries:
• FastMCP: High-level Python framework
• LiteLLM: Multi-model abstraction
• Prometheus: Metrics and monitoring

Community:
• MCP Discord - Architecture channel
• GitHub - MCP Examples repository
• Blog posts from enterprise adopters

✨ Final Thoughts

Expert MCP isn't about complexity for its own sake—it's about building systems that scale gracefully.

When you started, one server with a few tools seemed sufficient.
Now you understand that production systems need:

  • Smart context management to control costs
  • Multi-model support to optimize quality vs price
  • Discoverable tools so AI can find what it needs
  • Modular architecture that teams can maintain

The patterns you've learned apply far beyond MCP:

  • Context engineering → Resource optimization in any system
  • Multi-model servers → Polyglot architectures
  • Tool discovery → API design and developer experience
  • Naming conventions → System-wide consistency
Labels:

Comments

Post a Comment