Como dividir um pipeline Python em serviços MCP independentes

O caminho usual para construir um pipeline com múltiplas etapas é colocar tudo no mesmo processo e chamar as partes por função. Funciona — até que cada parte vira uma mini-aplicação com dependências, modos de falha e ciclos de lançamento próprios. É nesse ponto que compartilhar um processo deixa de ser conveniente e passa a ser o que quebra primeiro.

O problema do monólito

Imagine um orquestrador com três motores — extração, risco e fraude — instanciados dentro da mesma classe, no mesmo ambiente virtual. Três modos de falha se escondem aí:

• Uma exceção não tratada na extração derruba a pontuação de risco, porque não há fronteira entre elas;
• Um upgrade de dependência do motor de fraude pode quebrar a instalação do motor de extração;
• Corrigir um bug em qualquer serviço exige reimplantar a aplicação inteira.

Os problemas aparecem meses depois do lançamento, quando cada motor acumulou lógica e dependências reais. A falha costuma ser um “version pin” que força um downgrade silencioso em outro lugar, ou uma extração pesada que rouba memória do motor de risco no mesmo processo.

Fronteiras de processo, não apenas MCP

A correção de verdade não é o MCP em si, mas uma fronteira real de processo em torno de cada serviço — para que uma falha em um não afete os outros, uma mudança de dependência em um não respingue no outro, e cada um possa ser implantado e escalado no próprio ritmo. Você conseguiria isso com endpoints REST simples. O que o MCP adiciona é uma forma única e consistente de qualquer orquestrador ou agente descobrir o que um serviço faz e chamá-lo, sem escrever uma integração sob medida por consumidor.

Essa distinção evita usar MCP por inércia: se um serviço só tem um chamador e isso não vai mudar, uma API interna direta é mais simples. O caso do MCP é específico — um orquestrador que precisa alcançar um número crescente de serviços de forma uniforme.

Um servidor MCP por serviço

Com o FastMCP, escrever uma ferramenta é decorar uma função e tipar os argumentos. O interessante é o que você deixa de fora de cada servidor para que o isolamento seja real:

# extraction_server.py
from fastmcp import FastMCP
mcp = FastMCP("extraction-tools")

@mcp.tool
def extract_fields(document: str) -> dict:
    result = extraction_engine.run(document)
    if result.confidence < 0.6:
        raise ValueError("confidence below threshold")
    return result.fields

if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=8931)

Nenhum arquivo importa o outro; eles rodam como processos separados, em portas separadas. Se o motor de risco precisa de um upgrade de biblioteca no próximo trimestre, é uma mudança em um processo, testada e enviada no próprio cronograma.

O isolamento é uma disciplina

O ponto fácil de errar: a fronteira só se sustenta se você a respeitar ao escrever as ferramentas. É tentador importar um utils.py compartilhado nos dois servidores — mas isso recria exatamente o acoplamento que você criou dois servidores para evitar. Se ambos precisam da mesma lógica de validação, ela é duplicada deliberadamente ou vira um terceiro serviço.

Orquestração com LangGraph

O trabalho do orquestrador não é escolher uma ferramenta, e sim computar um caminho entre serviços, onde a saída de cada etapa vira a entrada da próxima. Com langchain-mcp-adapters e o StateGraph do LangGraph, dá para expressar esse caminho explicitamente — e, quando um terceiro ou quarto serviço entrar, trocar as arestas fixas por arestas condicionais.

Limitações que continuam em aberto

O isolamento resolve a dor original, mas deixa questões reais: o que acontece quando o caminho se revela errado no meio da execução, quando uma falha passa a ser uma chamada de rede em vez de uma exceção três frames acima, ou quando o orquestrador tem mais serviços do que uma única decisão consegue sustentar. São problemas separados — e continuam separados.