MÓDULO 3.5

🔌 MCP Avançado

Crie servidores MCP completos com FastMCP, adicione autenticação Bearer, faça deploy em Railway e explore o ecossistema de 200+ servidores prontos. MCP é a camada de interoperabilidade do ecossistema agentic.

6
Tópicos
45
Minutos
Inter.
Nível
Projeto
Tipo
1

🏛️ Revisão da arquitetura MCP

MCP (Model Context Protocol) é um protocolo client-server padronizado para comunicação entre agentes de IA e fontes de ferramentas/dados. Um agente (cliente MCP) descobre ferramentas disponíveis em um servidor MCP e as chama via JSON-RPC.

🌐 Fluxo completo MCP

1
Conexão:Agente conecta ao servidor via stdio (local) ou HTTP+SSE (remoto)
2
Descoberta:Agente chama tools/list — recebe lista de tools com nome, descrição e schema
3
Seleção:LLM escolhe qual tool chamar com base nas descrições e no contexto
4
Execução:Agente chama tools/call — servidor executa e retorna resultado

💡 Por que MCP é importante

Sem MCP, cada framework tem sua própria forma de definir tools — LangChain, CrewAI, Llama Index são todos incompatíveis. Com MCP, um servidor de ferramentas funciona com qualquer cliente compatível: LangGraph, CrewAI, Claude Desktop, Cursor.

MCP é o USB-C das ferramentas de IA — um padrão que conecta tudo.

2

🛠️ Criando servidor MCP completo com FastMCP

FastMCP é a biblioteca Python que torna criação de servidores MCP trivial. Você decora funções com @mcp.tool() e o servidor expõe automaticamente o tools/list e tools/call com schema gerado a partir dos type hints.

🔨 Servidor MCP completo

# servidor.py
from fastmcp import FastMCP
from pydantic import BaseModel
from typing import Optional
import httpx

# Inicializa o servidor
mcp = FastMCP(
    name="empresa-tools",
    description="Ferramentas internas da Empresa XYZ para agentes de IA"
)

# Tool 1: busca de clientes
@mcp.tool()
async def buscar_cliente(
    customer_id: str,
    include_history: bool = False
) -> dict:
    """
    Busca informações de um cliente pelo ID.

    Args:
        customer_id: ID único do cliente (formato: CUS_xxxxx)
        include_history: Se True, inclui histórico de compras

    Returns:
        Dict com nome, email, plano e histórico (opcional)
    """
    # Lógica real: consulta banco de dados
    cliente = await db.get_customer(customer_id)
    result = {
        "id": cliente.id,
        "nome": cliente.nome,
        "email": cliente.email,
        "plano": cliente.plano,
        "created_at": cliente.created_at.isoformat()
    }
    if include_history:
        result["compras"] = await db.get_purchases(customer_id)
    return result

# Tool 2: envio de email
@mcp.tool()
async def enviar_email(
    destinatario: str,
    assunto: str,
    corpo: str,
    template: Optional[str] = None
) -> dict:
    """
    Envia email para um destinatário usando templates corporativos.

    Returns:
        Dict com message_id e status de entrega
    """
    msg_id = await email_service.send(
        to=destinatario,
        subject=assunto,
        body=corpo,
        template=template
    )
    return {"message_id": msg_id, "status": "enviado"}

# Resource: acesso a dados estáticos
@mcp.resource("config://empresa/politicas")
async def get_politicas() -> str:
    """Retorna as políticas de atendimento da empresa em Markdown."""
    return open("politicas.md").read()

# Inicia o servidor
if __name__ == "__main__":
    mcp.run(transport="stdio")       # local
    # mcp.run(transport="sse", port=8080)  # HTTP

💡 Docstrings geram as descrições do LLM

As docstrings das suas funções se tornam as descrições das tools que o LLM vê. Uma docstring clara e específica faz o LLM escolher a tool certa no momento certo. Trate as docstrings como prompt engineering para seu servidor MCP.

3

🔒 Autenticação e segurança em servidores MCP

Servidores MCP acessíveis via HTTP precisam autenticar clientes. Bearer token no header Authorization é o padrão mais simples para uso interno. OAuth 2.0 para cenários com múltiplos usuários e escopos granulares.

🔐 Autenticação Bearer com FastAPI

# servidor_seguro.py
from fastmcp import FastMCP
from fastapi import FastAPI, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import os, hashlib, hmac

app = FastAPI()
security = HTTPBearer()
mcp = FastMCP(name="empresa-tools-seguro")

# Tokens válidos (em produção: banco de dados)
VALID_TOKENS = {
    os.environ["AGENTE_TOKEN_1"]: "agente-producao",
    os.environ["AGENTE_TOKEN_2"]: "agente-staging",
}

def validar_token(credentials: HTTPAuthorizationCredentials) -> str:
    """Valida o Bearer token e retorna o nome do cliente."""
    token = credentials.credentials
    cliente = VALID_TOKENS.get(token)
    if not cliente:
        raise HTTPException(status_code=401, detail="Token inválido")
    return cliente

# Middleware de autenticação
@app.middleware("http")
async def auth_middleware(request, call_next):
    if request.url.path.startswith("/mcp"):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return JSONResponse({"error": "Unauthorized"}, status_code=401)
        token = auth.removeprefix("Bearer ")
        if token not in VALID_TOKENS:
            return JSONResponse({"error": "Invalid token"}, status_code=401)
        # Log de auditoria
        logger.info(f"Tool call from {VALID_TOKENS[token]}")
    return await call_next(request)

# Rate limiting simples
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)

@app.get("/mcp/tools/list")
@limiter.limit("60/minute")  # 60 chamadas por minuto
async def list_tools(request: Request):
    return await mcp.handle_list_tools()

# Cliente configurando autenticação
# No LangGraph/CrewAI, passe o header na configuração do servidor MCP:
# MCPClient(url="https://...", headers={"Authorization": "Bearer TOKEN"})

⚠️ Checklist de segurança para MCP em produção

  • Nunca use tokens hardcoded no código — use variáveis de ambiente
  • Nunca exponha o servidor sem HTTPS em produção
  • Audite todas as chamadas com timestamp, cliente e parâmetros
  • Rate limit por cliente para prevenir abuso
  • Valide e sanitize todos os inputs antes de executar
4

🔭 Descoberta dinâmica de ferramentas

O protocolo MCP permite que agentes descubram ferramentas dinamicamente em runtime — sem precisar hardcodar as tools no código do agente. Adicione uma tool no servidor e todos os agentes a verão automaticamente na próxima conexão.

🔍 Exemplo completo de descoberta

# O que o servidor retorna em tools/list
{
  "tools": [
    {
      "name": "buscar_cliente",
      "description": "Busca informações de um cliente pelo ID...",
      "inputSchema": {
        "type": "object",
        "properties": {
          "customer_id": {"type": "string", "description": "ID único (CUS_xxxxx)"},
          "include_history": {"type": "boolean", "default": false}
        },
        "required": ["customer_id"]
      }
    },
    {
      "name": "enviar_email",
      "description": "Envia email para destinatário usando templates...",
      "inputSchema": { ... }
    }
  ]
}

# Como o LangGraph consome o servidor MCP
from langchain_mcp_adapters.client import MultiServerMCPClient

async def criar_agente_com_mcp():
    async with MultiServerMCPClient({
        "empresa": {
            "url": "https://mcp.empresa.com/sse",
            "transport": "sse",
            "headers": {"Authorization": f"Bearer {TOKEN}"}
        },
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "transport": "stdio"
        }
    }) as client:
        # Tools descobertas automaticamente dos dois servidores
        tools = client.get_tools()  # lista completa combinada

        llm = ChatOpenAI(model="gpt-4o")
        agente = llm.bind_tools(tools)

        # Agente pode usar tools dos dois servidores
        resposta = await agente.ainvoke("Liste os PRs abertos do repo X")
5

🚀 MCP em produção: deploy e operação

Um servidor MCP local só funciona em desenvolvimento. Para agentes em produção, o servidor precisa ser um serviço HTTP com alta disponibilidade. Railway, Fly.io e Docker são as opções mais usadas.

🐳 Dockerfile + Railway deploy

# Dockerfile
FROM python:3.11-slim
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Health check endpoint
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s \
  CMD curl -f http://localhost:8080/health || exit 1

EXPOSE 8080
CMD ["python", "servidor.py", "--transport", "sse", "--port", "8080"]

---
# requirements.txt
fastmcp>=0.9.0
fastapi>=0.109.0
uvicorn>=0.27.0
pydantic>=2.0.0
httpx>=0.26.0

---
# servidor.py (versão HTTP para produção)
from fastmcp import FastMCP
import uvicorn, os

mcp = FastMCP(name="empresa-tools")

# Health check endpoint
@mcp.app.get("/health")
async def health():
    return {"status": "ok", "version": "1.0.0"}

# Suas tools aqui...
@mcp.tool()
async def buscar_cliente(customer_id: str) -> dict:
    """Busca cliente pelo ID."""
    return await db.get_customer(customer_id)

if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    uvicorn.run(mcp.app, host="0.0.0.0", port=port)

---
# Deploy no Railway (após instalar Railway CLI)
# railway init
# railway up
# railway variables set AGENTE_TOKEN_1=... DATABASE_URL=...
# URL gerada: https://empresa-tools.railway.app
🚄

Railway

  • • Deploy em 5 minutos
  • • Auto-deploy via GitHub
  • • Free tier para teste
  • • HTTPS automático
✈️

Fly.io

  • • Escala para zero
  • • Deploy global multi-região
  • • Volumes persistentes
  • • Mais controle de infra
🐳

Docker / K8s

  • • Controle total
  • • Infra própria
  • • Compliance total
  • • Mais setup inicial
6

🌐 Ecossistema de servidores prontos

Existe um ecossistema crescente de mais de 200 servidores MCP prontos para uso. Antes de construir do zero, verifique se já existe. A maioria das integrações comuns já está disponível como servidor plug-and-play.

Servidores Oficiais Anthropic

  • filesystem — acesso ao sistema de arquivos local
  • github — repositórios, PRs, issues
  • postgres — consultas SQL seguras
  • brave-search — busca web com Brave API
  • slack — mensagens e canais
  • puppeteer — automação de browser

Servidores da Comunidade

  • notion — workspace e databases
  • linear — project management
  • stripe — pagamentos e assinaturas
  • jira — tickets e sprints
  • aws-kb — AWS Knowledge Base
  • playwright — testes E2E automatizados

Usando servidores prontos em 2 minutos

# Instalação (npx não requer instalação permanente)
# npx -y @modelcontextprotocol/server-github

# Configuração no claude_desktop_config.json
{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."
      }
    },
    "postgres": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres",
               "postgresql://localhost/mydb"]
    },
    "empresa": {
      "url": "https://mcp.empresa.com/sse",
      "headers": {"Authorization": "Bearer ..."}
    }
  }
}

# Descoberta em mcp.so (marketplace)
# → Visite mcp.so para explorar 200+ servidores
# → Filtre por categoria: databases, productivity, devtools
# → Cada servidor tem README com instalação e config

🎯 Resumo do Módulo

MCP = USB-C das ferramentas — um servidor funciona com qualquer cliente compatível
FastMCP — @mcp.tool() transforma funções Python em tools MCP automaticamente
Bearer token — autenticação simples para servidores HTTP internos
Deploy em Railway — Dockerfile + railway up = servidor MCP público em produção
200+ servidores prontos — GitHub, Postgres, Slack, Notion e muitos mais no mcp.so

Próximo Módulo:

3.6 — ⚖️ Comparativo de Frameworks: tabela de decisão completa e guia definitivo de quando usar cada um