moai-mcp-builder✓ تایید شده

محتوای SKILL.md

---
name: "moai-mcp-builder"
version: "4.0.0"
description: Enterprise MCP (Model Context Protocol) server development using FastMCP 2.0 with production-grade tools, resources, prompts, and intelligent agent-first design. Use when building MCP servers, integrating with LLMs, creating agent tools, implementing RAG systems, or developing protocol-based AI integration solutions.
allowed-tools: 
  - Read
  - Bash
  - WebSearch
  - WebFetch
status: stable
---

# Enterprise MCP Server Builder & AI Integration Platform

## MCP Protocol Capabilities

**Model Context Protocol (MCP)**:
- Standard protocol for LLM context provisioning
- Tool provisioning (agent-callable functions)
- Resource system (data/documents exposure)
- Prompt templates (multi-turn patterns)
- Transport abstraction (stdio, SSE, HTTP)

**FastMCP 2.0 Framework**:
- Python-first MCP implementation
- Type-safe decorators (@mcp.tool, @mcp.resource)
- Automatic OpenAPI generation
- Enterprise auth (OAuth, SAML)
- Proxy and composition patterns
- Production-ready deployments

## Skill Metadata
| Field | Value |
| ----- | ----- |
| **Version** | **4.0.0 Enterprise** |
| **Created** | 2025-11-12 |
| **Framework** | FastMCP 2.0, Python SDK |
| **Protocol** | Model Context Protocol (MCP) |
| **Features** | Tools, Resources, Prompts |
| **Transport** | Stdio, SSE, HTTP/WebSocket |
| **Auth** | OAuth 2.0, SAML, API Keys |
| **Tier** | **4 (Enterprise)** |

## MCP Architecture Overview

### Three Core Components

**1. Tools** (Agent-Callable Functions):
```
@mcp.tool
def search_documents(query: str) -> list[dict]:
    """Search documents by query"""
    # Implementation
    return results

# Tools are exposed as callable actions
# Agents decide when/how to invoke
# Return typed structured data
```

**2. Resources** (Data/Document Exposure):
```
@mcp.resource
def get_document(doc_id: str) -> str:
    """Fetch document content"""
    # Read from database
    return content

# Resources provide context without execution
# Large documents efficiently streamed
# Agents request as needed
```

**3. Prompts** (Multi-Turn Patterns):
```
@mcp.prompt("analyze-code")
def code_analysis(language: str, code_snippet: str) -> str:
    """System prompt for code analysis"""
    return f"""You are a {language} expert...
    Analyze this code:
    {code_snippet}"""

# Pre-built system prompts
# Contextual parameters
# Multi-turn conversation patterns
```

## Tool Design Principles

### Agent-Centric Design

**Principle 1: Build for Workflows, Not APIs**
```
BAD:
@mcp.tool
def create_event(title: str) -> dict: ...
@mcp.tool
def check_availability(date: str) -> bool: ...

GOOD:
@mcp.tool
def schedule_event(
    title: str,
    date: str,
    check_conflicts: bool = True
) -> dict:
    """Single tool combining availability check + creation"""
    if check_conflicts and has_conflict(date):
        raise ValueError(f"Conflict on {date}")
    return create_event(title, date)
```

**Principle 2: Optimize for Limited Context**
```
BAD:
def list_all_users() -> list[dict]:
    """Returns 10,000 users with all fields"""
    return all_users_with_full_data()

GOOD:
def search_users(
    query: str,
    limit: int = 10,
    fields: list[str] = ["id", "name", "email"]
) -> list[dict]:
    """Search with pagination and field filtering"""
    return paginated_filtered_search(query, limit, fields)
```

**Principle 3: Design Actionable Error Messages**
```
BAD:
raise ValueError("Invalid date")

GOOD:
raise ValueError(
    f"Date must be in future. "
    f"Current date: {today}. "
    f"Try: {(today + timedelta(days=1)).isoformat()}"
)
```

**Principle 4: Follow Natural Task Subdivisions**
```
Tool Naming Conventions:
- create_*: Create new resources
- update_*: Modify existing resources
- delete_*: Remove resources
- list_*: Enumerate resources
- get_*: Fetch specific resource
- search_*: Find by criteria
- analyze_*: Generate insights
```

### Type Safety & Documentation

**Parameter Types**:
```python
from typing import Literal, Optional, Union
from pydantic import Field, validator

@mcp.tool
def process_data(
    data: dict,
    mode: Literal["fast", "thorough"] = "fast",
    timeout: Optional[int] = None,
    callbacks: list[str] = Field(default_factory=list),
) -> dict:
    """
    Process data with specified mode.
    
    Args:
        data: Input dictionary
        mode: Processing speed preference
        timeout: Optional timeout in seconds
        callbacks: List of webhook URLs
        
    Returns:
        Processed result dictionary
    """
    # Implementation with validation
```

## FastMCP Server Implementation

### Basic Server Structure

```python
from fastmcp import FastMCP

server = FastMCP("my-server")

@server.tool()
def get_weather(city: str) -> str:
    """Get current weather for a city"""
    # Fetch from API
    return f"Weather in {city}: 72°F, sunny"

@server.resource("weather://{city}")
def weather_resource(city: str) -> str:
    """Get detailed weather data"""
    return f"""Weather Report for {city}
    Temperature: 72°F
    Humidity: 65%
    Wind: 10mph"""

@server.prompt("weather-analyst")
def weather_prompt() -> str:
    """System prompt for weather analysis"""
    return """You are a weather expert. Analyze the provided 
    weather data and give actionable recommendations."""

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

### Authentication Layer

```python
from fastmcp.auth import OAuth2Provider, APIKeyAuth

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

@server.auth(oauth)
@server.tool()
def protected_tool(user_id: str) -> dict:
    """Tool requiring OAuth authentication"""
    return get_user_data(user_id)

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

@server.auth(api_auth)
@server.resource("secure://{resource_id}")
def secure_resource(resource_id: str) -> str:
    """Resource with API key protection"""
    return fetch_secure_data(resource_id)
```

## Real-World Patterns

### Pattern 1: Database Query Tool

```python
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

engine = create_engine("postgresql://...")

@server.tool()
def query_database(
    table: Literal["users", "products", "orders"],
    filters: dict = None,
    limit: int = 10
) -> list[dict]:
    """
    Query database with filters and pagination.
    
    Examples:
    - table="users", filters={"status": "active"}
    - table="products", filters={"price_gt": 100}
    """
    with Session(engine) as session:
        query = get_query(table)
        if filters:
            query = apply_filters(query, filters)
        return [dict(row) for row in query.limit(limit)]
```

### Pattern 2: API Integration Tool

```python
import httpx
from datetime import datetime, timedelta

@server.tool()
def search_articles(
    query: str,
    source: Literal["hacker-news", "arxiv", "medium"] = "hacker-news",
    hours_back: int = 24
) -> list[dict]:
    """
    Search articles from multiple sources.
    
    Args:
        query: Search terms
        source: Publication source
        hours_back: Search time window
    """
    cutoff = datetime.now() - timedelta(hours=hours_back)
    
    clients = {
        "hacker-news": HNClient(),
        "arxiv": ArxivClient(),
        "medium": MediumClient(),
    }
    
    results = clients[source].search(query, since=cutoff)
    return sorted(results, key=lambda x: x["relevance"], reverse=True)[:10]
```

### Pattern 3: File Processing Tool

```python
from pathlib import Path
import json, yaml

@server.tool()
def process_config_file(
    file_path: str,
    format: Literal["json", "yaml", "toml"] = "json",
    validate: bool = True
) -> dict:
    """
    Load and validate configuration file.
    
    Args:
        file_path: Path to config file
        format: File format auto-detection
        validate: Validate against schema
    """
    path = Path(file_path)
    if not path.exists():
        raise FileNotFoundError(f"Config file not found: {file_path}")
    
    loaders = {
        "json": json.loads,
        "yaml": yaml.safe_load,
        "toml": tomli.loads,
    }
    
    content = path.read_text()
    config = loaders[format](content)
    
    if validate:
        validate_config_schema(config)
    
    return config
```

### Pattern 4: LLM Integration Tool

```python
from anthropic import Anthropic

client = Anthropic()

@server.tool()
def analyze_text(
    text: str,
    analysis_type: Literal["summary", "sentiment", "entities", "topics"],
    max_tokens: int = 1000
) -> str:
    """
    Analyze text using Claude API.
    
    Args:
        text: Input text to analyze
        analysis_type: Type of analysis
        max_tokens: Max response length
    """
    prompts = {
        "summary": f"Summarize: {text}",
        "sentiment": f"Analyze sentiment: {text}",
        "entities": f"Extract entities: {text}",
        "topics": f"Identify topics: {text}",
    }
    
    response = client.messages.create(
        model="claude-opus-4-1",
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompts[analysis_type]}]
    )
    
    return response.content[0].text
```

### Pattern 5: Proxy Server (Composing Multiple MCPs)

```python
from fastmcp.proxy import ProxyServer

proxy = ProxyServer("multi-protocol-proxy")

# Mount multiple MCP servers
proxy.mount("/weather", weather_server)
proxy.mount("/database", database_server)
proxy.mount("/api", api_server)

@proxy.tool()
def unified_search(
    query: str,
    search_in: Literal["weather", "database", "api"] = "all"
) -> dict:
    """
    Search across mounted servers.
    """
    results = {}
    if search_in in ["all", "weather"]:
        results["weather"] = proxy.invoke("/weather/search", query)
    if search_in in ["all", "database"]:
        results["db"] = proxy.invoke("/database/search", query)
    if search_in in ["all", "api"]:
        results["api"] = proxy.invoke("/api/search", query)
    return results

if __name__ == "__main__":
    proxy.run()
```

## Deployment Strategies

### Local Development

```bash
# Install dependencies
pip install fastmcp[dev]

# Run in development mode
fastmcp dev my_server.py

# Test with cURL
curl -X POST http://localhost:8000/tool/search_documents \
  -H "Content-Type: application/json" \
  -d '{"query": "python"}'
```

### Docker Deployment

```dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["fastmcp", "run", "server.py", "--host", "0.0.0.0"]
```

### Cloud Deployment

```yaml
# deploy.yaml (Kubernetes)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: mcp-server
        image: my-mcp-server:latest
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
```

## Best Practices

### Error Handling

```python
from fastmcp.exceptions import MCPError

@server.tool()
def resilient_tool(resource_id: str) -> dict:
    """Tool with comprehensive error handling"""
    try:
        resource = fetch_resource(resource_id)
        if not resource:
            raise MCPError(
                f"Resource not found: {resource_id}. "
                f"Try: list_available_resources()"
            )
        return process_resource(resource)
    except DatabaseError as e:
        raise MCPError(f"Database error: {str(e)}", retry=True)
    except ValidationError as e:
        raise MCPError(f"Invalid input: {e.details}", retry=False)
```

### Performance Optimization

```python
from functools import lru_cache
import asyncio

@lru_cache(maxsize=128)
def expensive_computation(key: str) -> dict:
    """Cache results of expensive operations"""
    return fetch_and_process(key)

@server.tool()
async def batch_process(items: list[str]) -> list[dict]:
    """Use async for concurrent operations"""
    tasks = [process_item_async(item) for item in items]
    return await asyncio.gather(*tasks)
```

### Monitoring & Logging

```python
import logging
from fastmcp.monitoring import instrument

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@server.tool()
@instrument(track_duration=True, track_errors=True)
def monitored_tool(data: dict) -> dict:
    """Tool with automatic monitoring"""
    logger.info(f"Processing: {data}")
    result = process(data)
    logger.info(f"Result: {result}")
    return result
```

---

**MCP Production Readiness Checklist**:
- [ ] All tools have type hints and docstrings
- [ ] Error messages are actionable
- [ ] Authentication configured (OAuth/API keys)
- [ ] Logging and monitoring enabled
- [ ] Rate limiting implemented
- [ ] Unit tests >85% coverage
- [ ] Integration tests for tool chains
- [ ] Performance benchmarks <500ms p99
- [ ] Documentation with examples
- [ ] Deployment strategy defined

مجوز

مجوز اعلام‌شده: MIT

MIT License

Copyright (c) 2026 majiayu000

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

مشاهدهٔ مجوز در مخزن منبع — نسخهٔ منتشرشده در آن‌جا مرجع است.