moai-connector-mcp

name	moai-connector-mcp
description	MCP 1.0+ Custom Server Development with FastMCP Framework
version	3.0.0
modularized	true
last_updated	"2025-11-30T00:00:00.000Z"
compliance_score	70
auto_trigger_keywords	["connector","mcp"]
color	red

Quick Reference

MCP Server Development Framework

What it does: Comprehensive guide to building, testing, and deploying custom MCP (Model Context Protocol) servers using FastMCP framework for exposing tools, resources, and prompts to Claude and other AI models.

Core Capabilities:

• ✅ FastMCP server development with type-safe decorators
• ✅ Tool/Resource/Prompt architecture patterns
• ✅ Pydantic validation and error handling
• ✅ OAuth2 & API Key authentication patterns
• ✅ Performance monitoring and health checks
• ✅ Docker & Kubernetes deployment
• ✅ Testing strategies and validation
• ✅ Production-grade patterns (caching, circuit breaker, rate limiting)

When to Use:

• Building custom MCP servers for internal tools
• Exposing existing services via MCP protocol
• Implementing enterprise authentication patterns
• Deploying MCP servers in production environments
• Optimizing server performance and reliability
• Testing MCP server implementations

---

Implementation Guide

Getting Started with FastMCP

Installation:

pip install fastmcp

Minimum Server:

from fastmcp import FastMCP

server = FastMCP("my-server")

@server.tool()
def hello_world(name: str) -> str:
    """Greet someone."""
    return f"Hello, {name}!"

if __name__ == "__main__":
    server.run()

Core Concepts:

1. Tools: Functions Claude can invoke with validated parameters
2. Resources: URI-based data endpoints for exposing information
3. Prompts: Reusable conversation templates and system prompts

Detailed guide: getting-started.md

---

MCP Server Architecture

Three-Component Pattern:

┌─────────────────────────────────────────┐
│         MCP Server (FastMCP)            │
├─────────────────────────────────────────┤
│  Tools (Functions)                      │
│  • @server.tool() decorator             │
│  • Pydantic validation                  │
│  • Workflow-optimized naming            │
│                                         │
│  Resources (Data Endpoints)             │
│  • @server.resource("uri://...") decor  │
│  • Streaming support                    │
│  • Permission-based access              │
│                                         │
│  Prompts (Templates)                    │
│  • @server.prompt("name") decorator     │
│  • Parameter injection                  │
│  • Multi-turn workflows                 │
└─────────────────────────────────────────┘
        ↓
    MCP Protocol (JSON-RPC 2.0)
        ↓
┌─────────────────────────────────────────┐
│    Claude / LLM Client                  │
└─────────────────────────────────────────┘

Design Patterns: server-design.md

---

Production-Ready Server Example

from fastmcp import FastMCP
from pydantic import Field
from typing import Literal, Optional

server = FastMCP("enterprise-database-server")

@server.tool()
def search_records(
    query: str,
    table: Literal["users", "products", "orders"],
    limit: int = Field(default=10, ge=1, le=100),
    filters: Optional[dict] = None
) -> dict:
    """
    Search database records with pagination.

    Args:
        query: Search query string
        table: Table to search
        limit: Max results (1-100)
        filters: Optional filter criteria

    Returns:
        Dict with results and metadata
    """
    if not query or not query.strip():
        raise ValueError("Query cannot be empty")

    results = execute_search(query, table, limit, filters)
    return {
        "status": "success",
        "count": len(results),
        "results": results,
        "total_available": get_total_count(query)
    }

@server.resource("db://{table}/{id}")
def get_record(table: , : ) -> :
    
    record = fetch_record(table, )
      record:
         ValueError()
     record

 __name__ == :
    server.run()

Implementation Guide: implementation.md

---

Authentication Patterns

OAuth2 (User-Authenticated):

from fastmcp.auth import OAuth2Provider

oauth = OAuth2Provider(
    authorize_url="https://auth.company.com/authorize",
    token_url="https://auth.company.com/token",
    scopes=["read:data", "write:data"]
)

@server.auth(oauth)
@server.tool()
def protected_action(user_id: str) -> dict:
    """Requires OAuth token."""
    return execute_action(user_id)

API Key (Service-to-Service):

from fastmcp.auth import APIKeyAuth

api_auth = APIKeyAuth(header="X-API-Key")

@server.auth(api_auth)
@server.resource("secure://{resource_id}")
def secure_resource(resource_id: str) -> str:
    """Requires API key."""
    return fetch_data(resource_id)

Detailed Patterns: auth-patterns.md

---

Testing MCP Servers

Unit Testing:

import pytest
from fastmcp import FastMCP

@pytest.fixture
def server():
    s = FastMCP("test-server")

    @s.tool()
    def add(a: int, b: int) -> int:
        return a + b

    return s

def test_add_tool(server):
    result = server.invoke_tool("add", {"a": 2, "b": 3})
    assert result == 5

def test_invalid_params(server):
    with pytest.raises(ValueError):
        server.invoke_tool("add", {"a": "not-a-number", "b": 3})

Testing Guide: testing.md

---

Deployment

Docker:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY server.py .
EXPOSE 8000
CMD ["python", "server.py"]

Kubernetes:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-server
  template:
    metadata:
      labels:
        app: mcp-server
    spec:
      containers:
      - name: mcp-server
        image: mcp-server:latest
        ports:
        - containerPort: 8000
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 10

Deployment Guide: deployment.md

---

Advanced Patterns

Tool Design Best Practices

✅ DO:

• Design tools for single meaningful workflow steps
• Validate all inputs with Pydantic Field constraints
• Provide clear, actionable error messages
• Use pagination for large result sets
• Include comprehensive docstrings

❌ DON'T:

• Mix multiple responsibilities in one tool
• Return unlimited result sets
• Skip input validation
• Expose sensitive data
• Ignore error handling

Tool Design Guide: tool-design.md

---

Performance Optimization

Caching Strategy:

from functools import wraps
from datetime import datetime, timedelta

class MCPCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds

    def cached(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = str((func.__name__, args, kwargs))
            if key in self.cache:
                value, timestamp = self.cache[key]
                if (datetime.now() - timestamp).total_seconds() < self.ttl:
                    return value

            result = func(*args, **kwargs)
            self.cache[key] = (result, datetime.now())
            return result
        return wrapper

cache = MCPCache(ttl_seconds=600)

@cache.cached
def expensive_operation(param: str) -> dict:
    return fetch_and_process(param)

Advanced Patterns:

• Caching
• Circuit Breaker
• Rate Limiting

---

Monitoring & Observability

Health Checks:

@server.resource("health://status")
def health_check() -> dict:
    """Server health status."""
    return {
        "status": "healthy",
        "version": "1.0.0",
        "uptime_seconds": get_uptime(),
        "active_connections": get_connection_count()
    }

Metrics & Logging:

import logging
import time

logger = logging.getLogger(__name__)

@server.tool()
def monitored_operation(params: dict) -> dict:
    start = time.time()
    try:
        result = execute_operation(params)
        duration = time.time() - start
        logger.info(f"Operation completed in {duration:.2f}s")
        return result
    except Exception as e:
        logger.error(f"Operation failed: {str(e)}")
        raise

Monitoring Guide: monitoring.md

---

Works Well With

• moai-context7-integration - Documentation access for API patterns
• moai-cc-configuration - MCP server configuration management
• moai-essentials-debug - Server debugging and troubleshooting
• moai-domain-backend - Backend service architecture
• moai-domain-cloud - Cloud deployment patterns
• moai-quality-security - Security validation and OWASP compliance

---

Core Concepts

1. FastMCP Framework: Python library for rapid MCP server development
2. Type Safety: Pydantic models ensure Claude understands parameter constraints
3. Workflow Design: Tools for single meaningful tasks, not granular APIs
4. Authentication Strategy: OAuth2 for user apps, API keys for service-to-service
5. Production Readiness: Monitoring, health checks, error handling, caching
6. Testing: Comprehensive test coverage before deployment

---

Module Navigation

Getting Started:

• getting-started.md - FastMCP basics

Core Development:

• server-design.md - Architecture patterns
• implementation.md - Real examples
• testing.md - Test strategies

Deployment & Operations:

• deployment.md - Docker/Kubernetes

Advanced Patterns (modules/development/patterns/):

• tool-design.md - Tool best practices
• resource-design.md - Resource patterns
• auth-patterns.md - Authentication
• monitoring.md - Observability
• caching.md - Cache strategies
• circuit-breaker.md - Resilience
• rate-limiting.md - Rate control

---

Changelog

• v3.0.0 (2025-11-27): Complete restructure from server integration to server development focus, modularized with development patterns
• v2.1.0 (2025-11-22): Modularized structure - SKILL.md refactored, reference.md and examples.md added
• v2.0.0 (2025-11-22): MCP 1.0+ protocol complete spec update
• v1.0.0 (2025-11-21): Initial MCP integration skill

---

Status: Production Ready | See modules/development/ for detailed patterns | Last Updated: 2025-11-27