Painel
PT
Comece grátis

Como criar um MCP server em Python para uma API de dados web

Um MCP server funcional em 109 linhas de Python: três tools de dados web, erros que dizem ao agente o que fazer e orçamento de créditos por sessão. Testado em 19/10/2026.

Para criar um MCP server em Python, instale o pacote oficial mcp (2.3.0 em 19 de outubro de 2026), crie um MCPServer, decore funções assíncronas com @mcp.tool() e chame mcp.run() para stdio. Um servidor que envolve uma API REST é uma camada fina: as type hints viram o schema de entrada, a docstring vira as instruções que o modelo lê, e um ToolError lançado vira um erro sobre o qual o agente pode agir.

Abaixo está um servidor completo de 109 linhas que envolve três endpoints do FastAF (busca, notícias, leitura de páginas), com validação de entrada, erros legíveis pelo agente, timeouts e um orçamento de créditos por sessão. Executei-o via stdio com o cliente do SDK e o registrei no Claude Code em 19 de outubro de 2026.

Qual pacote e qual import usar hoje?

Instale o mcp do PyPI (pip install mcp httpx). A versão 2.3.0 saiu em 2 de outubro de 2026 e exige Python 3.10 ou mais novo (PyPI). A grande mudança em relação à era 1.x: FastMCP foi renomeado para MCPServer, e from mcp.server.fastmcp import FastMCP agora lança ModuleNotFoundError. O novo import é from mcp.server.mcpserver import MCPServer (guia de migração). Tutoriais antigos ainda mostram o import velho.

Como é o servidor completo?

Salve como fastaf_web.py. Ele lê FASTAF_API_KEY e usa por padrão https://api.fastaf.co/v1; FASTAF_BASE_URL sobrescreve.

"""A small MCP server that wraps three FastAF endpoints. Run: python fastaf_web.py"""
import asyncio
import os
from typing import Annotated, Any

import httpx
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations
from pydantic import Field

BASE = os.environ.get("FASTAF_BASE_URL", "https://api.fastaf.co/v1")
BUDGET = int(os.environ.get("FASTAF_SESSION_BUDGET", "100"))  # credits per server process
COST = {"serp": 5, "news": 5, "scrape": 1}  # published list costs, used to reserve
spent = 0
lock = asyncio.Lock()

mcp = MCPServer("fastaf-web")
http = httpx.AsyncClient(
    base_url=BASE,
    headers={"Authorization": f"Bearer {os.environ['FASTAF_API_KEY']}"},
    timeout=httpx.Timeout(45.0, connect=5.0),
)
READ_ONLY = ToolAnnotations(read_only_hint=True, open_world_hint=True)


async def call(endpoint: str, **params: Any) -> httpx.Response:
    """GET an endpoint, reserve its cost in the session budget, turn failures into advice for the model."""
    global spent
    cost = COST[endpoint]
    async with lock:  # reserve before the await
        if spent + cost > BUDGET:
            raise ToolError(
                f"Session budget reached ({spent} of {BUDGET} credits used). Do not call "
                "more tools; answer from what you already have and say the budget ran out."
            )
        spent += cost
    for attempt in (1, 2):
        try:
            r = await http.get(f"/{endpoint}", params=params)
        except httpx.TimeoutException:
            raise ToolError("Timed out after 45 s. Retry once, then try a different query.")
        if r.status_code == 429 and attempt == 1:  # one polite retry, as the API asks
            await asyncio.sleep(min(float(r.headers.get("Retry-After", 2)), 10))
            continue
        break
    # API errors are never charged; a timeout keeps the reservation.
    spent += (int(r.headers.get("X-FastAF-Credits-Used", cost)) if r.status_code == 200 else 0) - cost
    if r.status_code == 200:
        return r
    code = r.json()["error"]["code"] if "json" in r.headers.get("content-type", "") else "unknown"
    if r.status_code == 402:
        advice = "The account is out of credits. Stop calling this server and tell the user to top up."
    elif r.status_code == 429:
        advice = "Rate limited even after one retry. Wait a few seconds, then call again, one request at a time."
    elif code == "blocked_target":
        advice = "That URL is blocked on purpose (private address or login page). Use a different public source."
    elif r.status_code == 400:
        advice = "The request was rejected. Check the arguments and try a simpler call."
    else:
        advice = "Temporary failure on the API side, not charged. Retry once."
    raise ToolError(f"{endpoint} failed: HTTP {r.status_code} {code}. {advice}")


@mcp.tool(annotations=READ_ONLY)
async def search_web(
    query: Annotated[str, Field(min_length=1, max_length=300, description="Plain search query, as typed into Google")],
    results: Annotated[int, Field(ge=1, le=10)] = 5,
) -> dict[str, Any]:
    """Search Google and return ranked results (title, url, snippet). Costs 5 credits per call.
    Use it to find URLs; it does not return page content. Follow up with read_page on the
    one or two most relevant URLs instead of searching again with a near-identical query."""
    r = await call("serp", query=query, results=results)
    hits = [{"rank": o["rank"], "title": o["title"], "url": o["link"], "snippet": o.get("snippet", "")}
            for o in r.json().get("organic_results", [])[:results]]
    return {"results": hits, "session_credits_used": spent}


@mcp.tool(annotations=READ_ONLY)
async def search_news(
    query: Annotated[str, Field(min_length=1, max_length=300, description="Topic or entity to find recent news about")],
    results: Annotated[int, Field(ge=1, le=10)] = 5,
) -> dict[str, Any]:
    """Search Google News for recent articles (title, source, age, url, snippet). Costs 5
    credits per call. Use it for anything that happened in the last weeks; use search_web
    for documentation and evergreen pages."""
    r = await call("news", query=query, results=results)
    items = [{"title": n["title"], "source": n.get("source"), "age": n.get("lastUpdated"),
              "url": n["url"], "snippet": n.get("snippet", "")}
             for n in r.json().get("news_results", [])[:results]]
    return {"results": items, "session_credits_used": spent}


@mcp.tool(annotations=READ_ONLY)
async def read_page(
    url: Annotated[str, Field(pattern=r"^https?://", max_length=2000, description="Public http(s) page URL")],
    max_chars: Annotated[int, Field(ge=1000, le=30000, description="Character budget for the returned text")] = 8000,
) -> dict[str, Any]:
    """Fetch a public web page and return it as markdown. Costs 1 credit. If 'truncated' is true,
    only the first max_chars characters came back: raise max_chars (up to 30000) only if
    the part you need is missing. Cannot read pages behind logins or paywalls."""
    r = await call("scrape", url=url, formats="markdown")
    text = r.text
    return {"url": url, "text": text[:max_chars], "truncated": len(text) > max_chars,
            "total_chars": len(text), "session_credits_used": spent}


if __name__ == "__main__":
    mcp.run()  # stdio; never print() to stdout here, it carries the protocol

Três escolhas carregam o peso. Cada tool devolve um dict enxuto (o endpoint de notícias inclui thumbnails em base64, que o modelo nunca deveria ver). O read_page recebe um orçamento de caracteres e informa se cortou a página. E toda docstring declara o custo e quando usar a tool em vez das vizinhas, porque esse texto é toda a documentação que o modelo recebe.

Como uma tool MCP deve informar erros ao modelo?

Lance ToolError com uma mensagem que diga o que fazer em seguida. O SDK a devolve com is_error=True, então o modelo lê o seu texto; qualquer outra exceção é tratada como falha e o modelo vê apenas Error executing tool <name> (tratamento de erros). Devolver uma string de erro é a armadilha: ela chega como um resultado bem-sucedido.

A especificação traça a mesma linha: erros de execução de tool carregam “feedback acionável que modelos de linguagem podem usar para se autocorrigir” (especificação de tools); erros de protocolo são para requisições que o modelo não consegue corrigir.

O helper call() mapeia os erros do FastAF para instruções: 402 insufficient_credits significa parar e avisar o usuário (tentar de novo não ajuda); 429 espera o Retry-After uma vez e depois manda o agente desacelerar; 400 blocked_target diz que a URL está bloqueada de propósito, então procure outra fonte pública. Requisições que falham nunca são cobradas. O Pydantic rejeita argumentos ruins (Field(ge=1, le=10), o padrão ^https?://) antes de qualquer requisição HTTP ser feita.

Como limitar os créditos por sessão?

Antes de cada chamada, sob um asyncio.Lock, confira spent + cost > BUDGET com o custo de tabela (5 para busca e notícias, 1 para um scrape) e some o custo ao contador antes do primeiro await, para que chamadas paralelas não passem as duas. Se não couber, lance um ToolError que mande o agente parar e responder com o que já tem. Depois da resposta, troque a reserva pelo header X-FastAF-Credits-Used: um erro da API não é cobrado e a devolve, um timeout do cliente a mantém. Um servidor stdio vive enquanto o cliente o mantiver, então um processo é uma sessão.

Dois limites. O contador zera se o cliente reiniciar o servidor (grave-o em um arquivo para persistir). E é uma proteção por sessão, não um limite da conta: o saldo de créditos pertence à conta e é compartilhado por todas as suas chaves. O post sobre o teto de gastos cobre as outras camadas.

O que ele fez quando eu o executei?

Iniciei o servidor a partir do cliente do SDK (Client(StdioServerParameters(...))) e chamei cada tool em 19 de outubro de 2026:

Chamada Resultado Tempo Créditos
search_web, 3 resultados title, url e snippet ranqueados 2,5 a 3,2 s (três execuções) 5
read_page no guia de migração do SDK, max_chars=4000 4.000 de 223.596 caracteres, truncated: true 0,9 a 1,1 s 1
search_news, 3 resultados title, source, age, url, snippet 1,8 a 9,0 s (três execuções) 5
read_page em http://192.168.1.1/ blocked_target com orientação menos de 0,1 s 0
results=50, ou URL ftp:// rejeitado pela validação, sem chamada HTTP 0,0 s 0
orçamento definido em 6, segunda busca recusada, sem chamada HTTP; uma leitura de página de 1 crédito ainda passou 0,0 s 0

Um servidor stub que retornava 402 e 429 produziu as mensagens esperadas, e um 429 seguido de 200 teve sucesso na nova tentativa em 1,03 s (uma simulação, não a API real). O lock veio depois das execuções reais e foi verificado só com um stub: com orçamento de 6, duas buscas paralelas deixam uma passar.

Uma execução de confirmação de search_web e read_page pelo servidor, na API de produção, custou 6 créditos: 9,6 s para a busca, 0,7 s para a página. Na primeira tentativa a API respondeu 429 duas vezes seguidas, o servidor devolveu a orientação de “limite de taxa” em vez de quebrar, e a nova execução passou.

Como registrá-lo no Claude Code e no Cursor?

Coloque o servidor em um .mcp.json do projeto. O Claude Code expande ${VAR} dentro de env, então o arquivo não contém nenhum segredo (documentação de MCP do Claude Code, conferida em 19 de outubro de 2026):

{
  "mcpServers": {
    "fastaf-web": {
      "type": "stdio",
      "command": "/path/to/venv/bin/python",
      "args": ["/path/to/fastaf_web.py"],
      "env": {
        "FASTAF_API_KEY": "${FASTAF_API_KEY}",
        "FASTAF_SESSION_BUDGET": "50"
      }
    }
  }
}

A forma pela CLI é claude mcp add --transport stdio --env FASTAF_API_KEY=... fastaf-web -- /path/to/venv/bin/python /path/to/fastaf_web.py, que guarda a chave em ~/.claude.json. Com o .mcp.json acima, o Claude Code 2.1.197 em modo headless chamou search_news e devolveu os títulos.

O Cursor lê .cursor/mcp.json ou ~/.cursor/mcp.json com o mesmo formato e resolve ${env:FASTAF_API_KEY} (documentação do Cursor, conferida em 19 de outubro de 2026). Não executei o Cursor para este post.

Vale construir o seu próprio servidor ou usar o hospedado?

Se você só quer as tools no seu agente, use o servidor hospedado: uma URL e um header Bearer, 15 tools, sem código para manter. O post sobre a configuração hospedada cobre isso, e a documentação do MCP lista cada tool.

Construa o seu quando quiser um conjunto enxuto de tools (três tools custam menos contexto ao modelo que quinze), cache ou log próprios, prompts ajustados ao seu domínio, um orçamento imposto no seu processo ou composição com tools internas. O mesmo wrapper serve para qualquer outro endpoint: veja as páginas da API de busca e da API de web scraping para os parâmetros, a documentação para o contrato completo e os preços para os custos.

Resumindo

Instale o mcp 2.3.0, use MCPServer (não FastMCP), lance ToolError com orientação para o agente, limite caracteres e créditos dentro do servidor e registre-o com uma entrada stdio no .mcp.json. Pegue uma chave com 500 créditos grátis e rode o servidor acima.