Construyendo Tu Primer Servidor MCP: Guia de Produccion para Ingenieros Backend

from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP("deploy-tools")
 
@mcp.tool()
def get_deploy_status(service_name: str, environment: str) -> dict:
    """Get the current deployment status for a service.
 
    Args:
        service_name: The name of the service (e.g., 'api-gateway', 'auth-service')
        environment: The target environment ('staging' or 'production')
 
    Returns:
        A dict with keys: service, environment, status, version, deployed_at
    """
    # In reality, this calls your deployment API
    status = deployment_api.get_status(service_name, environment)
    return {
        "service": service_name,
        "environment": environment,
        "status": status.state,
        "version": status.version,
        "deployed_at": status.timestamp.isoformat(),
    }
 
if __name__ == "__main__":
    mcp.run(transport="stdio")

import re
from enum import Enum
from pydantic import BaseModel, field_validator
 
class Environment(str, Enum):
    STAGING = "staging"
    PRODUCTION = "production"
 
class DeployStatusRequest(BaseModel):
    service_name: str
    environment: Environment
 
    @field_validator("service_name")
    @classmethod
    def validate_service_name(cls, v: str) -> str:
        if not re.match(r"^[a-z][a-z0-9\-]{1,62}[a-z0-9]$", v):
            raise ValueError(
                "service_name must be lowercase alphanumeric with hyphens, "
                "3-64 characters"
            )
        return v
 
@mcp.tool()
def get_deploy_status(service_name: str, environment: str) -> dict:
    """Get the current deployment status for a service.
 
    Args:
        service_name: Lowercase alphanumeric with hyphens (e.g., 'api-gateway')
        environment: Must be 'staging' or 'production'
    """
    req = DeployStatusRequest(
        service_name=service_name,
        environment=environment,
    )
    status = deployment_api.get_status(req.service_name, req.environment.value)
    return {
        "service": req.service_name,
        "environment": req.environment.value,
        "status": status.state,
        "version": status.version,
        "deployed_at": status.timestamp.isoformat(),
    }

import logging
from mcp.server.fastmcp import FastMCP, Context
 
logger = logging.getLogger(__name__)
 
class ToolError(Exception):
    def __init__(self, message: str, retryable: bool = False):
        self.message = message
        self.retryable = retryable
        super().__init__(message)
 
@mcp.tool()
async def get_deploy_status(
    service_name: str,
    environment: str,
    ctx: Context,
) -> dict:
    """Get the current deployment status for a service."""
    try:
        req = DeployStatusRequest(
            service_name=service_name,
            environment=environment,
        )
    except ValueError as e:
        return {
            "error": str(e),
            "retryable": False,
            "hint": "Check service_name format and environment values",
        }
 
    try:
        status = await deployment_api.get_status(
            req.service_name, req.environment.value
        )
    except deployment_api.ServiceNotFound:
        return {
            "error": f"Service '{req.service_name}' not found",
            "retryable": False,
            "hint": "Use list_services tool to see available services",
        }
    except deployment_api.APITimeout:
        logger.warning(
            "Deployment API timeout for %s/%s",
            req.service_name,
            req.environment.value,
        )
        await ctx.report_progress(0, 1, "Deployment API is slow, retrying...")
        return {
            "error": "Deployment API timed out",
            "retryable": True,
            "hint": "The deployment API is experiencing delays. Try again.",
        }
    except Exception:
        logger.exception("Unexpected error in get_deploy_status")
        return {
            "error": "Internal server error",
            "retryable": False,
            "hint": "Contact the platform team if this persists",
        }
 
    return {
        "service": req.service_name,
        "environment": req.environment.value,
        "status": status.state,
        "version": status.version,
        "deployed_at": status.timestamp.isoformat(),
    }

import os
import jwt
from functools import wraps
from datetime import datetime, timezone
 
JWT_SECRET = os.environ["MCP_JWT_SECRET"]
ALLOWED_SCOPES = {"deploy:read", "deploy:write", "services:list"}
 
def verify_token(token: str) -> dict:
    """Verify JWT and return claims. Raises on invalid tokens."""
    try:
        payload = jwt.decode(
            token,
            JWT_SECRET,
            algorithms=["HS256"],
            options={"require": ["exp", "sub", "scopes"]},
        )
    except jwt.ExpiredSignatureError:
        raise ToolError("Token expired", retryable=False)
    except jwt.InvalidTokenError as e:
        raise ToolError(f"Invalid token: {e}", retryable=False)
 
    return payload
 
def require_scope(scope: str):
    """Decorator to enforce scope-based authorization on tools."""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            ctx = kwargs.get("ctx")
            if ctx is None:
                raise ToolError("Missing context", retryable=False)
 
            token = ctx.request_context.get("auth_token", "")
            claims = verify_token(token)
            user_scopes = set(claims.get("scopes", []))
 
            if scope not in user_scopes:
                raise ToolError(
                    f"Insufficient permissions. Required: {scope}",
                    retryable=False,
                )
 
            kwargs["_claims"] = claims
            return await func(*args, **kwargs)
        return wrapper
    return decorator
 
@mcp.tool()
@require_scope("deploy:read")
async def get_deploy_status(
    service_name: str,
    environment: str,
    ctx: Context,
    _claims: dict = None,
) -> dict:
    """Get deployment status. Requires deploy:read scope."""
    # _claims contains the verified JWT payload
    logger.info(
        "deploy_status_check user=%s service=%s env=%s",
        _claims["sub"],
        service_name,
        environment,
    )
    # ... rest of implementation

import time
import uuid
import logging
import structlog
 
logger = structlog.get_logger()
 
def with_observability(func):
    """Wrap tool calls with structured logging and timing."""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        request_id = str(uuid.uuid4())[:8]
        tool_name = func.__name__
        start = time.monotonic()
 
        log = logger.bind(
            request_id=request_id,
            tool=tool_name,
            params={
                k: v for k, v in kwargs.items()
                if k not in ("ctx", "_claims")
            },
        )
        log.info("tool_invoked")
 
        try:
            result = await func(*args, **kwargs)
            elapsed = time.monotonic() - start
            log.info(
                "tool_completed",
                duration_ms=round(elapsed * 1000, 2),
                has_error="error" in result if isinstance(result, dict) else False,
            )
            return result
        except Exception as e:
            elapsed = time.monotonic() - start
            log.error(
                "tool_failed",
                duration_ms=round(elapsed * 1000, 2),
                error=str(e),
            )
            raise
    return wrapper
 
@mcp.tool()
@with_observability
async def get_deploy_status(service_name: str, environment: str, ctx: Context) -> dict:
    """Get the current deployment status for a service."""
    # ... implementation

import pytest
from mcp.server.fastmcp import FastMCP
 
@pytest.fixture
def mcp_server():
    server = FastMCP("test-deploy-tools")
 
    @server.tool()
    async def get_deploy_status(service_name: str, environment: str) -> dict:
        if service_name == "nonexistent":
            return {"error": "Service not found", "retryable": False}
        return {
            "service": service_name,
            "environment": environment,
            "status": "running",
            "version": "1.2.3",
        }
 
    return server
 
@pytest.mark.anyio
async def test_valid_deploy_status(mcp_server):
    async with mcp_server.test_client() as client:
        result = await client.call_tool(
            "get_deploy_status",
            {"service_name": "api-gateway", "environment": "staging"},
        )
        assert result[0].text
        data = json.loads(result[0].text)
        assert data["status"] == "running"
        assert data["service"] == "api-gateway"
 
@pytest.mark.anyio
async def test_invalid_service_name(mcp_server):
    async with mcp_server.test_client() as client:
        result = await client.call_tool(
            "get_deploy_status",
            {"service_name": "nonexistent", "environment": "staging"},
        )
        data = json.loads(result[0].text)
        assert "error" in data
        assert data["retryable"] is False
 
@pytest.mark.anyio
async def test_tool_listing(mcp_server):
    async with mcp_server.test_client() as client:
        tools = await client.list_tools()
        tool_names = [t.name for t in tools]
        assert "get_deploy_status" in tool_names

import time
from collections import defaultdict
 
class RateLimiter:
    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window = window_seconds
        self.calls: dict[str, list[float]] = defaultdict(list)
 
    def check(self, key: str) -> bool:
        now = time.monotonic()
        window_start = now - self.window
        self.calls[key] = [t for t in self.calls[key] if t > window_start]
        if len(self.calls[key]) >= self.max_calls:
            return False
        self.calls[key].append(now)
        return True
 
rate_limiter = RateLimiter(max_calls=30, window_seconds=60)
 
@mcp.tool()
async def get_deploy_status(service_name: str, environment: str, ctx: Context) -> dict:
    """Get deployment status. Rate limited to 30 calls per minute."""
    caller = ctx.request_context.get("client_id", "anonymous")
    if not rate_limiter.check(caller):
        return {
            "error": "Rate limit exceeded. Max 30 calls per minute.",
            "retryable": True,
            "hint": "Wait before retrying. Consider batching requests.",
            "retry_after_seconds": 60,
        }
    # ... rest of implementation

from mcp.server.fastmcp import FastMCP
 
mcp = FastMCP(
    "deploy-tools",
    host="0.0.0.0",
    port=8080,
)
 
# Register all your tools, resources, prompts...
 
if __name__ == "__main__":
    mcp.run(transport="streamable-http")