🏗️ Construindo um MCP server próprio

→ {"jsonrpc":"2.0","id":1,"method":"initialize",
   "params":{"protocolVersion":"2024-11-05","capabilities":{}}}
← {"jsonrpc":"2.0","id":1,"result":{"serverInfo":{"name":"acme-crm","version":"0.3.1"},
   "capabilities":{"tools":{},"resources":{}}}}

→ {"jsonrpc":"2.0","id":2,"method":"tools/list"}
← {"jsonrpc":"2.0","id":2,"result":{"tools":[
    {"name":"search_orders","description":"...","inputSchema":{...}},
    {"name":"get_user",      "description":"...","inputSchema":{...}},
    {"name":"send_email",    "description":"...","inputSchema":{...}}
  ]}}

→ {"jsonrpc":"2.0","id":3,"method":"tools/call",
   "params":{"name":"search_orders","arguments":{"customer_id":"42","status":"open"}}}
← {"jsonrpc":"2.0","id":3,"result":{"content":[
    {"type":"text","text":"3 orders found: #1041, #1056, #1098"}
  ]}}

# servers/acme-crm/server.py
import os
from typing import Literal
from fastmcp import FastMCP
import httpx

mcp = FastMCP("acme-crm")

CRM_BASE  = os.environ["ACME_CRM_BASE_URL"]
CRM_TOKEN = os.environ["ACME_CRM_TOKEN"]

_client = httpx.Client(
    base_url=CRM_BASE,
    headers={"Authorization": f"Bearer {CRM_TOKEN}"},
    timeout=10.0,
)

@mcp.tool()
def search_orders(
    customer_id: str,
    status: Literal["open", "paid", "shipped", "all"] = "open",
    limit: int = 20,
) -> list[dict]:
    """Search orders for a given customer. Returns at most `limit` rows."""
    r = _client.get(
        "/v2/orders",
        params={"customer": customer_id, "status": status, "limit": limit},
    )
    r.raise_for_status()
    return r.json()["data"]

@mcp.tool()
def get_user(user_id: str) -> dict:
    """Fetch full user profile (name, email, tier, lifetime_value)."""
    r = _client.get(f"/v2/users/{user_id}")
    r.raise_for_status()
    return r.json()

@mcp.tool()
def send_email(
    to: str,
    subject: str,
    body_markdown: str,
    template: str | None = None,
) -> dict:
    """Send a transactional email via the CRM mailer. Idempotent by subject+to within 24h."""
    r = _client.post("/v2/email/send", json={
        "to": to, "subject": subject,
        "body": body_markdown, "template": template,
    })
    r.raise_for_status()
    return {"message_id": r.json()["id"], "queued": True}

if __name__ == "__main__":
    mcp.run()   # default: stdio transport

$ pip install fastmcp httpx
$ ACME_CRM_BASE_URL=https://crm.acme.internal \
  ACME_CRM_TOKEN=sk_test_xxx \
  npx @modelcontextprotocol/inspector python server.py

# Abre UI em http://localhost:5173 — lista tools, executa call, mostra JSON-RPC bruto

// servers/acme-crm/server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

const CRM_BASE  = process.env.ACME_CRM_BASE_URL!;
const CRM_TOKEN = process.env.ACME_CRM_TOKEN!;

async function crm(path: string, init: RequestInit = {}) {
  const r = await fetch(`${CRM_BASE}${path}`, {
    ...init,
    headers: { Authorization: `Bearer ${CRM_TOKEN}`, "content-type": "application/json", ...(init.headers || {}) },
  });
  if (!r.ok) throw new Error(`CRM ${r.status}: ${await r.text()}`);
  return r.json();
}

const tools = {
  search_orders: {
    description: "Search orders for a given customer.",
    schema: z.object({
      customer_id: z.string(),
      status: z.enum(["open", "paid", "shipped", "all"]).default("open"),
      limit: z.number().int().min(1).max(100).default(20),
    }),
    handler: async (a: any) => crm(`/v2/orders?customer=${a.customer_id}&status=${a.status}&limit=${a.limit}`),
  },
  get_user: {
    description: "Fetch full user profile.",
    schema: z.object({ user_id: z.string() }),
    handler: async (a: any) => crm(`/v2/users/${a.user_id}`),
  },
  send_email: {
    description: "Send transactional email. Idempotent by subject+to within 24h.",
    schema: z.object({
      to: z.string().email(),
      subject: z.string(),
      body_markdown: z.string(),
      template: z.string().optional(),
    }),
    handler: async (a: any) => crm("/v2/email/send", {
      method: "POST",
      body: JSON.stringify({ to: a.to, subject: a.subject, body: a.body_markdown, template: a.template }),
    }),
  },
};

const server = new Server({ name: "acme-crm", version: "0.3.1" }, { capabilities: { tools: {} } });

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: Object.entries(tools).map(([name, t]) => ({
    name, description: t.description, inputSchema: zodToJsonSchema(t.schema),
  })),
}));

server.setRequestHandler(CallToolRequestSchema, async (req) => {
  const t = (tools as any)[req.params.name];
  if (!t) throw new Error(`unknown tool ${req.params.name}`);
  const args = t.schema.parse(req.params.arguments ?? {});
  const result = await t.handler(args);
  return { content: [{ type: "text", text: JSON.stringify(result, null, 2) }] };
});

await server.connect(new StdioServerTransport());

# ~/.hermes/config.yaml
mcpServers:
  acme-crm:
    command: python
    args: ["/opt/mcp/acme-crm/server.py"]
    env:
      ACME_CRM_BASE_URL: https://crm.acme.internal
      ACME_CRM_TOKEN: ${secrets.ACME_CRM_TOKEN}   # resolve de ~/.hermes/secrets

  # ou versão Node
  acme-crm-node:
    command: node
    args: ["/opt/mcp/acme-crm/dist/server.js"]
    env:
      ACME_CRM_BASE_URL: https://crm.acme.internal
      ACME_CRM_TOKEN: ${secrets.ACME_CRM_TOKEN}

# Versão JWT por usuário — Hermes passa user_email no contexto
from fastmcp import FastMCP, Context
import jwt, time, os

mcp = FastMCP("acme-crm")
SIGNING_KEY = os.environ["ACME_JWT_PRIVATE_KEY"]   # PEM, carregado de secret

def mint_jwt(user_email: str) -> str:
    return jwt.encode(
        {
            "sub": user_email,
            "iss": "hermes-mcp",
            "aud": "acme-crm",
            "iat": int(time.time()),
            "exp": int(time.time()) + 300,   # 5 min
        },
        SIGNING_KEY,
        algorithm="RS256",
    )

@mcp.tool()
def get_user(ctx: Context, user_id: str) -> dict:
    """Fetch user — runs AS the calling agent's user, audited downstream."""
    token = mint_jwt(ctx.session.user_email)
    r = httpx.get(
        f"{CRM_BASE}/v2/users/{user_id}",
        headers={"Authorization": f"Bearer {token}"},
        timeout=10.0,
    )
    r.raise_for_status()
    return r.json()

# servers/acme-crm/Dockerfile
FROM python:3.12-slim
WORKDIR /app

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

COPY server.py .

# stdio mode: Hermes spawns this as subprocess via docker run -i
# SSE mode: expose 8080 e troca para mcp.run(transport="sse")
ENV PYTHONUNBUFFERED=1
ENTRYPOINT ["python", "server.py"]