Katana MCP Server - Comprehensive Architecture Design¶

Date: 2025-10-29 Version: 1.0 Status: Reference Document

⚠️ CURRENT IMPLEMENTATION: This document provides comprehensive background on MCP best practices and architectural patterns. For the current implementation plan, see MCP_V0.1.0_IMPLEMENTATION_PLAN.md.

Executive Summary¶

This document outlines a comprehensive redesign of the Katana MCP server based on:

2025 MCP best practices from the official specification
Modern architectural patterns for production MCP servers
Deep understanding of Katana Manufacturing ERP capabilities
Real-world manufacturing workflows and use cases

Key Design Principles¶

Single Responsibility: One MCP server focused on Katana ERP
Defense in Depth: Layered security with validation and sanitization
Fail-Safe Design: Graceful degradation under failure
Production Excellence: Observability, monitoring, health checks
User-Centric: Design around actual manufacturing workflows

MCP Primitives & Best Practices¶

Available MCP Features¶

Primitive	Purpose	When to Use
Tools	Functions AI can execute	Actions, computations, external system interactions
Resources	Context/data for AI	Expose existing information, documents, datasets
Prompts	Templated workflows	Guide user interactions, common patterns
Sampling	Server-initiated LLM requests	Autonomous workflows, multi-step reasoning
Roots	URI/filesystem boundaries	Data source/location discovery
Elicitation	Request user input	Clarification, confirmation, additional data

2025 Best Practices (from modelcontextprotocol.info)¶

Architecture¶

✅ Single Responsibility - One clear purpose per server
✅ Defense in Depth - Network isolation, auth, authorization, validation
✅ Fail-Safe Design - Circuit breakers, caching, rate limiting, safe defaults

Implementation¶

✅ Configuration Management - Environment variables, validation, secrets
✅ Comprehensive Error Handling - 4xx client, 5xx server, external errors
✅ Performance Optimization - Connection pooling, multi-level caching, async

Production Operations¶

✅ Monitoring & Observability - Structured logging, metrics, tracing
✅ Health Checks - Database, cache, APIs, disk, memory
✅ Deployment Strategies - Rolling updates, resource limits, autoscaling

Performance Targets (from best practices)¶

Metric	Target
Throughput	>1000 req/sec per instance
P95 Latency	\<100ms (simple ops)
P99 Latency	\<500ms (complex ops)
Error Rate	\<0.1%
Availability	>99.9%

Katana API Capabilities¶

Core Domains (52 API endpoints organized)¶

1. Catalog Management (Products, Materials, Services)¶

Products (finished goods): 8 endpoints
Materials (raw materials): 8 endpoints
Services (external): 8 endpoints
Variants (SKU-level): 8 endpoints
BOMs (recipes): 7 endpoints

2. Inventory Operations¶

Inventory levels: 7 endpoints
Stock adjustments: 6 endpoints
Stock transfers: 7 endpoints
Stocktakes (counts): 6 endpoints
Batches & serial numbers: 10 endpoints
Storage bins & locations: 9 endpoints

3. Order Management¶

Sales orders: 9 endpoints
Purchase orders: 8 endpoints
Manufacturing orders: 10 endpoints
Sales returns: 8 endpoints
Order fulfillments: 7 endpoints

4. Business Relations¶

Customers: 6 endpoints
Suppliers: 6 endpoints
Addresses: 12 endpoints

5. Configuration & Admin¶

Price lists: 22 endpoints
Tax rates: 4 endpoints
Custom fields: 3 endpoints
Webhooks: 8 endpoints
Users: 3 endpoints
Factories & locations: 8 endpoints

Manufacturing Workflow Patterns¶

┌────────────────────────────────────────────────────────────────┐
│                    MANUFACTURING WORKFLOW                       │
└────────────────────────────────────────────────────────────────┘

1. CATALOG SETUP
   ├─ Create products with variants
   ├─ Define materials and BOMs
   └─ Set up suppliers and pricing

2. SALES PROCESS
   ├─ Receive sales order
   ├─ Check inventory availability
   ├─ Create manufacturing orders if needed
   └─ Fulfill and ship

3. PROCUREMENT
   ├─ Monitor low stock
   ├─ Create purchase orders
   ├─ Receive and inspect goods
   └─ Update inventory

4. PRODUCTION
   ├─ Schedule manufacturing orders
   ├─ Allocate materials
   ├─ Track production progress
   ├─ Perform quality checks
   └─ Complete and stock finished goods

5. INVENTORY MANAGEMENT
   ├─ Track stock levels
   ├─ Perform stock counts
   ├─ Transfer between locations
   └─ Adjust for discrepancies

Proposed MCP Architecture¶

High-Level Structure¶

┌─────────────────────────────────────────────────────────────┐
│                    KATANA MCP SERVER                         │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │    TOOLS     │  │  RESOURCES   │  │   PROMPTS    │      │
│  │              │  │              │  │              │      │
│  │ • Inventory  │  │ • Dashboard  │  │ • Workflows  │      │
│  │ • Orders     │  │ • Reports    │  │ • Templates  │      │
│  │ • Production │  │ • Analytics  │  │ • Guides     │      │
│  │ • Catalog    │  │ • Insights   │  │              │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
│                                                              │
├─────────────────────────────────────────────────────────────┤
│                   DOMAIN MODELS LAYER                        │
│  KatanaVariant • KatanaProduct • KatanaMaterial             │
│  KatanaSalesOrder • KatanaPurchaseOrder • KatanaMO          │
├─────────────────────────────────────────────────────────────┤
│                   RESILIENT CLIENT                           │
│  • Automatic retries  • Rate limiting  • Pagination         │
│  • Circuit breakers   • Caching        • Error handling     │
├─────────────────────────────────────────────────────────────┤
│                   KATANA API (76+ endpoints)                 │
└─────────────────────────────────────────────────────────────┘

Tool Categories (20-25 tools total)¶

A. Inventory & Catalog (7-8 tools)¶

search_products - Find products/materials/services
get_variant_details - Get full variant info with stock
check_stock_levels - Check availability across locations
list_low_stock - Find items needing reorder
adjust_stock - Manual stock adjustments
transfer_stock - Move between locations
search_batches - Find by batch/serial numbers
get_bom - Get recipe/bill of materials

B. Sales Orders (5-6 tools)¶

create_sales_order - New customer order
get_sales_order - Retrieve order details
list_sales_orders - Find orders (filtered)
update_sales_order_status - Change status
fulfill_sales_order - Mark as shipped
create_sales_return - Handle returns

C. Purchase Orders (4-5 tools)¶

create_purchase_order - Order from supplier
get_purchase_order - Retrieve PO details
list_purchase_orders - Find POs (filtered)
receive_purchase_order - Goods receipt
update_purchase_order - Modify existing PO

D. Manufacturing Orders (4-5 tools)¶

create_manufacturing_order - Schedule production
get_manufacturing_order - Retrieve MO details
list_manufacturing_orders - Find MOs (filtered)
start_manufacturing_order - Begin production
complete_manufacturing_order - Finish and stock

E. Business Relations (2-3 tools)¶

search_customers - Find customer records
search_suppliers - Find supplier records
get_customer_orders - Order history

Resource Categories (5-7 resources)¶

Resources expose read-only data for context.

A. Dashboard Resources¶

katana://dashboard/inventory - Current inventory summary
katana://dashboard/orders - Active orders overview
katana://dashboard/production - Manufacturing status

B. Report Resources¶

katana://reports/low-stock - Items below threshold
katana://reports/overdue-orders - Late orders
katana://reports/production-schedule - Upcoming MOs

C. Analytics Resources¶

katana://analytics/turnover - Inventory turnover rates
katana://analytics/lead-times - Supplier performance

Prompt Categories (8-12 prompts)¶

Prompts provide templated workflows for common tasks.

A. Inventory Management¶

inventory_check - "Check stock and suggest reorders"
receive_shipment - "Process incoming goods"
cycle_count - "Perform stock count"

B. Order Processing¶

new_sales_order - "Create and validate new order"
fulfill_order - "Pick, pack, ship workflow"
rush_order - "Expedite production for urgent order"

C. Production Planning¶

plan_production - "Schedule MOs based on demand"
material_requirements - "Calculate material needs"
production_start - "Begin manufacturing process"

D. Troubleshooting¶

investigate_shortage - "Find cause of stock discrepancy"
late_order_analysis - "Diagnose delays"
quality_issue - "Handle production defect"

Tools Design¶

Design Principles for Tools¶

Clear Single Purpose - Each tool does one thing well
Strict Input Validation - JSON schema with field validation
Comprehensive Error Handling - Client/server/external errors
Idempotency - Safe to retry (where possible)
Explicit Confirmation - For state-changing operations
Informative Responses - Include context for next steps

Tool Template Structure¶

class ToolRequest(BaseModel):
    """Strictly validated input schema."""
    field: str = Field(..., description="Clear description", min_length=1)

class ToolResponse(BaseModel):
    """Structured, informative output."""
    result: ResultType
    metadata: dict[str, Any] = Field(default_factory=dict)  # Context
    next_actions: list[str] = Field(default_factory=list)   # Suggestions

async def tool_impl(request: ToolRequest, context: Context) -> ToolResponse:
    """Implementation with error handling and logging."""
    logger.info("tool_started", **request.dict())

    try:
        # Input validation
        if not request.field:
            raise ValueError("Field required")

        # Business logic
        client = context.request_context.lifespan_context.client
        result = await client.domain.operation(request.field)

        # Success logging
        logger.info("tool_completed", result_count=len(result))

        return ToolResponse(
            result=result,
            metadata={"source": "katana_api"},
            next_actions=["Check result", "Take next step"]
        )

    except ValueError as e:
        # Client error (4xx)
        logger.warning("tool_validation_failed", error=str(e))
        raise
    except Exception as e:
        # Server error (5xx)
        logger.error("tool_failed", error=str(e), exc_info=True)
        raise

Example: Create Sales Order Tool (Enhanced)¶

class CreateSalesOrderRequest(BaseModel):
    """Request to create a new sales order."""
    customer_id: int = Field(..., description="Customer ID", gt=0)
    items: list[OrderItem] = Field(..., description="Order line items", min_items=1)
    notes: str | None = Field(None, description="Optional order notes")
    priority: Literal["normal", "high", "urgent"] = Field("normal")
    requested_delivery_date: date | None = None

    # Elicitation: Ask for confirmation before creating
    confirm: bool = Field(
        False,
        description="Set to true to confirm order creation"
    )

class OrderItem(BaseModel):
    """Line item in sales order."""
    variant_id: int = Field(..., gt=0)
    quantity: float = Field(..., gt=0)
    unit_price: float | None = None  # Auto-fill from price list

class CreateSalesOrderResponse(BaseModel):
    """Response with created order details."""
    order_id: int
    order_number: str
    total_amount: float
    estimated_delivery: date | None
    manufacturing_required: bool
    warnings: list[str] = Field(default_factory=list)
    next_actions: list[str] = Field(default_factory=list)

async def create_sales_order(
    request: CreateSalesOrderRequest,
    context: Context
) -> CreateSalesOrderResponse:
    """Create a new sales order with validation and confirmation.

    This tool implements best practices:
    - Input validation (customer exists, items valid)
    - Stock availability check
    - Automatic pricing from price lists
    - Manufacturing requirement detection
    - Explicit confirmation for creation
    - Informative warnings and next actions
    """
    logger.info("create_sales_order_started", customer_id=request.customer_id)

    client = context.request_context.lifespan_context.client

    # 1. Validate customer exists
    customer = await client.customers.get(request.customer_id)
    if not customer:
        raise ValueError(f"Customer {request.customer_id} not found")

    # 2. Validate items and check stock
    warnings = []
    for item in request.items:
        variant = await client.variants.get(item.variant_id)
        if not variant:
            raise ValueError(f"Variant {item.variant_id} not found")

        # Check stock availability
        stock = await client.inventory.check_stock(variant.sku)
        if stock.available < item.quantity:
            warnings.append(
                f"Insufficient stock for {variant.sku}: "
                f"need {item.quantity}, have {stock.available}"
            )

    # 3. Elicitation: Require confirmation if not provided
    if not request.confirm:
        # Return preview without creating
        return CreateSalesOrderResponse(
            order_id=0,  # Not created yet
            order_number="PREVIEW",
            total_amount=0.0,  # Calculate preview
            estimated_delivery=request.requested_delivery_date,
            manufacturing_required=len(warnings) > 0,
            warnings=warnings + [
                "⚠️ Order not created. Set confirm=true to create."
            ],
            next_actions=[
                "Review warnings",
                "Set confirm=true to proceed",
                "Or adjust quantities"
            ]
        )

    # 4. Create order
    order = await client.sales_orders.create(
        customer_id=request.customer_id,
        items=[{
            "variant_id": item.variant_id,
            "quantity": item.quantity,
        } for item in request.items],
        notes=request.notes,
    )

    logger.info("sales_order_created", order_id=order.id)

    return CreateSalesOrderResponse(
        order_id=order.id,
        order_number=order.order_number,
        total_amount=order.total_amount,
        estimated_delivery=order.estimated_delivery,
        manufacturing_required=len(warnings) > 0,
        warnings=warnings,
        next_actions=[
            f"View order: get_sales_order(order_id={order.id})",
            "Create manufacturing orders if needed",
            "Process payment",
        ]
    )

Resources Design¶

Resources expose read-only contextual data that LLMs can reference.

Design Principles¶

URI-based addressing - katana://domain/resource
Cached & efficient - Don't hit API on every access
Structured data - JSON/YAML for easy parsing
Time-bounded - Include timestamps, refresh info
Actionable - Link to relevant tools

Example: Inventory Dashboard Resource¶

@mcp.resource("katana://dashboard/inventory")
async def inventory_dashboard(context: Context) -> Resource:
    """Current inventory status dashboard.

    Provides:
    - Total SKU count
    - Low stock items (< threshold)
    - Out of stock items
    - Top movers (by turnover)
    - Slow movers
    - Stock value

    Refreshes: Every 5 minutes
    """
    client = context.request_context.lifespan_context.client

    # Get cached dashboard data
    dashboard = await client.inventory.get_dashboard(cache_ttl=300)

    return Resource(
        uri="katana://dashboard/inventory",
        mimeType="application/json",
        text=json.dumps({
            "generated_at": datetime.now().isoformat(),
            "next_refresh": (datetime.now() + timedelta(minutes=5)).isoformat(),
            "summary": {
                "total_skus": dashboard.total_skus,
                "total_value": dashboard.total_value,
                "low_stock_count": len(dashboard.low_stock),
                "out_of_stock_count": len(dashboard.out_of_stock),
            },
            "low_stock_items": [
                {
                    "sku": item.sku,
                    "name": item.name,
                    "current": item.current_stock,
                    "threshold": item.reorder_point,
                    "suggested_order_qty": item.economic_order_quantity,
                }
                for item in dashboard.low_stock[:10]
            ],
            "out_of_stock": [
                {"sku": item.sku, "name": item.name}
                for item in dashboard.out_of_stock[:10]
            ],
            "next_actions": [
                "Review low stock items",
                "Use list_low_stock tool for full list",
                "Create purchase orders for critical items",
            ]
        }, indent=2)
    )

Prompts Design¶

Prompts guide users through common workflows with structured templates.

Design Principles¶

Workflow-oriented - Match real business processes
Step-by-step - Clear progression
Context-aware - Reference current state
Interactive - Elicit clarifications
Tool-integrated - Call appropriate tools

Example: Fulfill Sales Order Workflow¶

@mcp.prompt("fulfill_order")
async def fulfill_order_prompt(context: Context, order_id: int) -> Prompt:
    """Guide user through order fulfillment process.

    Workflow:
    1. Retrieve order details
    2. Check inventory availability
    3. Allocate stock
    4. Generate pick list
    5. Confirm picking
    6. Generate packing slip
    7. Confirm shipping
    8. Update order status
    """
    return Prompt(
        name="fulfill_order",
        description=f"Fulfill sales order #{order_id}",
        messages=[
            PromptMessage(
                role="user",
                content=f"""I need to fulfill sales order #{order_id}.

Please help me through the process:

1. First, show me the order details and check if we have sufficient stock
2. If stock is available, generate a pick list
3. After I confirm picking, generate a packing slip
4. Finally, mark the order as shipped

Let's start by retrieving the order."""
            ),
            PromptMessage(
                role="assistant",
                content=f"""I'll help you fulfill order #{order_id}. Let me start by retrieving the order details and checking stock availability.

{{% call_tool name="get_sales_order" args={{"order_id": {order_id}}} %}}"""
            )
        ]
    )

Security & Production Readiness¶

Security Layers¶

1. Network Isolation¶

# Bind to localhost only
server.bind("127.0.0.1", 8080)

2. Authentication¶

# API key from environment
api_key = os.getenv("KATANA_API_KEY")
if not api_key:
    raise ValueError("KATANA_API_KEY required")

3. Authorization¶

# Tool-level permissions (future)
@mcp.tool(requires_permission="inventory:write")
async def adjust_stock(...):
    pass

4. Input Validation¶

# Pydantic strict mode
class StrictRequest(BaseModel):
    class Config:
        extra = "forbid"  # Reject unknown fields
        str_strip_whitespace = True
        min_anystr_length = 1

5. Output Sanitization¶

# Remove sensitive data
def sanitize_response(data: dict) -> dict:
    """Remove API keys, internal IDs, etc."""
    return {
        k: v for k, v in data.items()
        if k not in ["api_key", "internal_id"]
    }

6. Monitoring¶

# Structured logging with security events
logger.warning(
    "unauthorized_access_attempt",
    user=user_id,
    resource=resource_name,
    ip=request.ip
)

Production Checklist¶

Phase 1: Core Compliance¶

All tools have input validation
All tools have error handling
All tools have structured logging
Health check endpoint implemented
Configuration validated at startup

Phase 2: Security¶

Phase 3: Performance¶

Connection pooling (5-20 connections)
Multi-level caching (memory, Redis)
Async processing for long operations
Resource limits configured

Phase 4: Observability¶

Phase 5: Reliability¶

Circuit breakers on external calls
Graceful degradation modes
Health checks for dependencies
Rolling deployment strategy
Horizontal autoscaling

Implementation Phases¶

Phase 1: Foundation (Week 1-2)¶

Goal: Core tools with production patterns

Deliverables¶

Enhanced inventory tools (3 tools)
search_products - With caching, ranking
check_stock_levels - Multi-location support
list_low_stock - With reorder suggestions
Structured logging system
Health check endpoint
Input validation patterns
Error handling framework

Success Criteria¶

All tools have comprehensive error handling
Structured logs capture key events
P95 latency < 100ms

Phase 2: Core Tools (Week 3-4)¶

Goal: Essential order management

Deliverables¶

Sales order tools (4 tools)
Create, get, list, fulfill
Purchase order tools (3 tools)
Create, get, receive
Manufacturing order tools (3 tools)
Create, get, list
Elicitation for confirmations
Response metadata & next actions

Success Criteria¶

End-to-end order workflows functional
Confirmation required for state changes
All tools return actionable next steps

Phase 3: Resources & Prompts (Week 5)¶

Goal: Context and guided workflows

Deliverables¶

Dashboard resources (3 resources)
Inventory, orders, production
Report resources (2 resources)
Low stock, overdue orders
Workflow prompts (5 prompts)
New order, fulfillment, production
Resource caching layer
Prompt templates

Success Criteria¶

Resources refresh efficiently (\<1s)
Prompts guide complete workflows
Users can complete tasks without API knowledge

Phase 4: Advanced Features (Week 6-7)¶

Goal: Production excellence

Deliverables¶

Advanced inventory tools
Stock transfers, adjustments, batch tracking
Analytics resources
Turnover, lead times
Troubleshooting prompts
Shortage investigation, late orders
Performance optimization
Connection pooling, caching
Monitoring & alerting

Success Criteria¶

1000 req/sec throughput
\<0.1% error rate
Full observability stack

Phase 5: Polish & Documentation (Week 8)¶

Goal: Production ready

Deliverables¶

Success Criteria¶

99.9% availability in staging
All use cases documented
Security review passed

Success Metrics¶

User Experience Metrics¶

Metric	Target	Rationale
Task Completion Rate	>90%	Users successfully complete workflows
Time to Complete Order	\<2 min	From order creation to confirmation
Error Recovery Rate	>95%	Users recover from errors without support
Tool Discovery Time	\<30 sec	Users find the right tool quickly

Technical Metrics¶

Metric	Target	Rationale
API Call Success Rate	>99.5%	Including retries
P95 Response Time	\<100ms	Simple operations
P99 Response Time	\<500ms	Complex operations
Cache Hit Rate	>80%	For frequent queries
Availability	>99.9%	Production uptime

Business Metrics¶

Metric	Target	Rationale
Orders Processed/Day	Baseline	Track adoption
Automation Rate	>50%	LLM completes without human
Support Tickets	\<5/week	Measure usability
User Satisfaction	>4.5/5	NPS survey

Conclusion¶

This architecture provides:

Comprehensive Coverage - 20-25 tools covering all major workflows
Production Ready - Security, observability, performance
User Friendly - Prompts, resources, clear documentation
Maintainable - Clean patterns, single responsibility
Scalable - Connection pooling, caching, autoscaling

The phased approach allows us to:

Validate patterns early (Phase 1)
Deliver value incrementally (Phase 2-3)
Optimize for production (Phase 4)
Launch confidently (Phase 5)

Next Steps: Review this design, get feedback, and begin Phase 1 implementation.