Showing Posts From

Arquitetura

Agentic MLOps: como A2A e MCP estão substituindo DAGs do Airflow por equipes de agentes

Se você já manteve um pipeline de ML em produção com Airflow, sabe o que é acordar às 3h da manhã porque uma DAG de retraining falhou no step 14 de 23. O log diz Task failed: validation_step_3. Qual validation? De qual modelo? Com quais dados? Boa sorte. O artigo da InfoQ publicado em março de 2026 — "Architecting Agentic MLOps with A2A and MCP" — propõe algo que venho testando nos últimos meses: trocar DAGs rígidos por equipes de agentes que se comunicam via protocolos padronizados. Não é hype. É uma mudança de arquitetura com trade-offs reais que vale a pena entender. O problema com DAGs de MLOps Pipelines tradicionais de ML — Airflow, Prefect, Dagster — tratam MLOps como uma sequência linear: ingestão → feature engineering → treino → validação → deploy → monitoramento. Cada step é um nó no grafo. A lógica de decisão ("o modelo passou no threshold?", "precisa de rollback?") vira um emaranhado de BranchPythonOperator e XComs que ninguém quer debugar. O problema não é o Airflow. É que ML pipelines não são lineares. Validação pode exigir retreino com dados diferentes. Deploy pode precisar de canary progressivo com rollback automático. Monitoramento pode detectar drift e disparar retraining sem esperar o schedule. Tentar expressar isso como um DAG estático é como tentar desenhar um fluxograma para uma conversa — funciona no PowerPoint, quebra na realidade. A2A + MCP: os dois protocolos que habilitam a mudança Antes de entrar na arquitetura, vale alinhar os protocolos. Já cobri MCP em detalhe no post anterior, mas o resumo rápido:MCP (Model Context Protocol, Anthropic): protocolo de conexão entre agentes e ferramentas externas. O agente declara o que precisa, o MCP server expõe as capabilities. Pense nele como a interface entre o agente e o mundo — registries de modelo, buckets S3, APIs de monitoramento, o que for.A2A (Agent-to-Agent, Google): protocolo de comunicação entre agentes. Diferente do MCP que conecta agente→ferramenta, o A2A conecta agente→agente. Cada agente publica um Agent Card declarando suas capabilities, aceita Tasks via JSON-RPC, e pode negociar formatos de resposta. É o que permite que um Validation Agent peça ao Training Agent para retreinar com parâmetros específicos sem hardcodar essa lógica.A convergência dos dois é o que torna Agentic MLOps viável. MCP para acessar infraestrutura, A2A para coordenar decisões entre agentes. A arquitetura em camadas O paper da InfoQ propõe três agentes core: Orchestrator Agent O cérebro do pipeline. Recebe o trigger (schedule, webhook, drift alert) e decide o plano de execução. Diferente de uma DAG, o plano é dinâmico — o orchestrator avalia o contexto (qual modelo, qual dataset, qual o estado do último deploy) e monta a sequência em runtime. Validation Agent Responsável por qualidade do modelo. Roda suítes de teste, verifica drift de dados, compara métricas com baselines. O ponto-chave: via A2A, ele pode rejeitar um modelo e pedir retreino com instruções específicas ("accuracy caiu 3pp no segmento X, retreinar com oversampling desse segmento"). Em uma DAG, isso seria um loop com estado compartilhado que ninguém quer manter. Deployment Agent Gerencia canary, blue-green, rollback. Conecta via MCP ao Kubernetes, ao registry de modelos, ao Prometheus. Se o canary falha, comunica via A2A ao Orchestrator que decide o próximo passo — rollback, retreino, ou escalar para um humano. Hands-on: esqueleto de um pipeline agêntico Para materializar a ideia, montei um esqueleto usando CrewAI (que já suporta A2A e MCP nativamente desde a v0.8) com MCP servers para acessar MLflow e Kubernetes: # agentic_mlops_crew.yaml agents: orchestrator: role: "ML Pipeline Orchestrator" goal: "Coordinate model retraining and deployment" tools: - mcp_server: "mlflow-registry" # MCP: acessa model registry - mcp_server: "s3-datasets" # MCP: acessa datasets a2a_capabilities: - "plan_execution" - "escalation" validator: role: "Model Quality Gate" goal: "Validate model performance against baselines" tools: - mcp_server: "mlflow-registry" - mcp_server: "evidently-monitoring" # MCP: drift detection a2a_capabilities: - "validation_report" - "retrain_request" deployer: role: "Model Deployment Manager" goal: "Safe progressive rollout with automatic rollback" tools: - mcp_server: "k8s-serving" # MCP: KServe/Seldon - mcp_server: "prometheus-metrics" a2a_capabilities: - "canary_status" - "rollback_trigger"O fluxo em pseudo-código: # orchestrator recebe trigger trigger = await orchestrator.receive_task(event)# monta plano dinâmico baseado no contexto plan = orchestrator.plan( model=trigger.model_id, reason=trigger.reason, # "scheduled" | "drift_detected" | "manual" last_deployment=await mlflow.get_latest(trigger.model_id) )# treina e envia para validação via A2A model_artifact = await orchestrator.execute_training(plan) validation = await validator.validate( # A2A call model=model_artifact, baseline=plan.baseline_metrics, required_segments=plan.critical_segments )if validation.status == "REJECTED": # validator pode pedir retreino com instruções específicas plan = orchestrator.replan(validation.feedback) # loop controlado pelo orchestrator, não por uma DAG elif validation.status == "APPROVED": deployment = await deployer.canary_deploy( # A2A call model=model_artifact, traffic_pct=10, monitor_minutes=30 )A diferença fundamental: a lógica de decisão vive nos agentes, não no grafo. Quando o validator rejeita um modelo, ele não apenas retorna False — ele retorna contexto ("accuracy no segmento enterprise caiu 4pp, dataset de treino tem 12% menos amostras desse segmento vs. mês passado"). O orchestrator usa esse contexto para replanejar. Trade-offs reais: quando NÃO migrar Seria desonesto vender isso como solução universal. Aqui estão os trade-offs que encontrei:Aspecto DAG tradicional Agentic MLOpsLatência de decisão Milissegundos (if/else) Segundos (LLM inference por decisão)Custo Compute do step Compute + tokens de LLM por agenteDebuggability Log linear, fácil de rastrear Traces distribuídos, precisa de observabilidade sériaDeterminismo 100% reproduzível Decisões do LLM podem variar entre runsComplexidade inicial Alta (DAG), mas conhecida Alta (agentes), e poucos dominamO custo de LLM inference em cada decisão é real. Em um pipeline que roda 50 vezes por dia, cada chamada ao orchestrator com contexto de 4K tokens custa. Fiz a conta para um cenário com 3 agentes, 8 chamadas LLM por run, usando Claude Sonnet: **$2.40/dia** vs. zero de compute decisório no Airflow. Para pipelines de alta frequência, isso escala. E o determinismo é a objeção mais séria. Se o Validation Agent aprova um modelo na segunda-feira e rejeita o mesmo modelo na terça com os mesmos dados, você tem um problema de auditoria. A mitigação que funciona: usar LLMs com temperature 0 para decisões binárias e logar o chain-of-thought completo como artefato de compliance. Quando faz sentido migrar Na minha experiência, Agentic MLOps compensa quando:Seu pipeline tem lógica de decisão complexa — múltiplos caminhos de retreino, rollback condicional, validação por segmento Você já tem MCP servers para sua infra (MLflow, K8s, monitoramento) — montar isso do zero é um projeto separado A frequência do pipeline é baixa/média — diário ou semanal, não a cada 5 minutos Você precisa de feedback loops que hoje são manuais — o Validation Agent substitui aquele Slack alert que um engenheiro olha (ou não) antes de aprovar o deploySe seu pipeline é treino → valida threshold → deploy sem ramificações, Airflow resolve. Não complique. O que vem pela frente O paper da InfoQ menciona Agent Registries — um catálogo onde agentes de MLOps publicam suas capabilities via A2A e podem ser compostos dinamicamente. Imagine um marketplace interno onde o time de ML publica um "Feature Quality Agent" e o time de infra publica um "Cost Optimization Agent", e o orchestrator compõe os dois no mesmo pipeline sem ninguém escrever glue code. Ainda está cedo. A maioria das empresas não tem nem MCP servers para a infra de ML, muito menos agentes A2A em produção. Mas a direção é clara: MLOps vai de orquestração imperativa para coordenação declarativa. De DAGs para equipes. Se você já tem MCP rodando e está pensando no próximo passo, o repo de referência da InfoQ é um bom ponto de partida. E se você ainda está no Airflow com 47 BranchPythonOperators aninhados — bom, pelo menos agora sabe que existe alternativa.

MCP em produção — 97M downloads, design patterns do arxiv, e o que ainda quebra

Em novembro de 2024, quando a Anthropic lançou o Model Context Protocol, os SDKs tinham 2 milhões de downloads mensais e a maioria das pessoas nem sabia o que era. Eu lembro de olhar a spec e pensar "isso é interessante, mas quem vai adotar?". Dezesseis meses depois, são 97 milhões de downloads mensais e eu preciso admitir que estava errado. MCP virou o protocolo padrão de conexão entre LLMs e ferramentas externas. Claude, GPT-5.4, Gemini — todos suportam. São 5.800+ servers no ecossistema. 4.750% de crescimento. E agora saiu um paper no arxiv que finalmente documenta o que funciona e o que não funciona quando você tenta colocar isso em produção. O paper: arxiv 2603.13417 O paper "Bridging Protocol and Production: Design Patterns for Deploying AI Agents with MCP" é exatamente o tipo de documento que faltava. Não é um paper teórico sobre a beleza do protocolo — é um catálogo de design patterns extraídos de deploys reais de agentes usando MCP. Os patterns que mais me interessaram: 1. Gateway Pattern Em vez de cada agente se conectar diretamente a N MCP servers, você coloca um gateway na frente que gerencia conexões, auth e rate limiting. Parece óbvio, mas 90% dos tutoriais mostram conexão direta. Em produção com 10+ servers, sem gateway você vai ter um pesadelo de configuração e debug. 2. Tool Composition Pattern Combinar ferramentas de múltiplos MCP servers em uma única chamada do agente. O paper mostra que a melhor abordagem é composição declarativa — o agente declara o que precisa, e o orquestrador resolve quais servers chamar. Tentativas de composição imperativa (o agente decidindo a sequência de chamadas) são frágeis e difíceis de debugar. 3. Fallback Chain Pattern Quando um MCP server não responde, ter uma cadeia de fallback com servers alternativos. O paper documenta três estratégias: retry simples, fallback para server alternativo, e degradação graceful (retornar resultado parcial). Na prática, já implementei a terceira e é a que menos frustra o usuário final. 4. Context Window Management O pattern mais técnico e mais útil. MCP servers podem retornar quantidades enormes de contexto — um server de database pode devolver milhares de linhas. O paper propõe um context budget por tool call, onde o orquestrador limita o output de cada server para caber no context window do modelo. Sem isso, um server guloso come o contexto inteiro e o agente perde acesso às outras ferramentas. Hands-on: criando um MCP server Chega de teoria. Vamos montar um MCP server básico que expõe uma API de busca para um agente. TypeScript (SDK oficial) npm install @modelcontextprotocol/sdkimport { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { z } from "zod";const server = new McpServer({ name: "search-server", version: "1.0.0", });// Registra uma tool que o agente pode chamar server.tool( "search_docs", "Busca na documentação interna", { query: z.string().describe("Termo de busca"), limit: z.number().default(5).describe("Máximo de resultados"), }, async ({ query, limit }) => { // Aqui vai sua lógica real — Elasticsearch, pgvector, whatever const results = await searchIndex(query, limit); return { content: [ { type: "text", text: JSON.stringify(results, null, 2), }, ], }; } );const transport = new StdioServerTransport(); await server.connect(transport);Python (SDK via PyPI) pip install mcpfrom mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import TextContent, Toolserver = Server("search-server")@server.tool() async def search_docs(query: str, limit: int = 5) -> list[TextContent]: """Busca na documentação interna.""" results = await search_index(query, limit) return [TextContent(type="text", text=str(results))]async def main(): async with stdio_server() as (read, write): await server.run(read, write, server.create_initialization_options())if __name__ == "__main__": import asyncio asyncio.run(main())Ambos os SDKs usam stdio como transport padrão — o client spawna o server como processo filho e se comunica via stdin/stdout. Isso é simples para desenvolvimento, mas em produção você vai querer HTTP/SSE (Server-Sent Events), que ambos os SDKs já suportam. Conectando no Claude Desktop (teste rápido) Edite o claude_desktop_config.json: { "mcpServers": { "search-docs": { "command": "node", "args": ["./dist/server.js"] } } }Reinicie o Claude Desktop e a tool search_docs aparece disponível. O agente pode invocá-la naturalmente durante a conversa. O que ainda quebra em produção Aqui é onde eu troco o chapéu de entusiasta pelo de engenheiro cansado. MCP em produção tem problemas reais que o hype esconde. Auth cross-server O maior gap. Cada MCP server gerencia sua própria autenticação. Se você tem 15 servers, o usuário precisa autenticar em cada um separadamente. Não existe um padrão de SSO ou token federation nativo no protocolo. O blog da WorkOS documenta bem esse problema e propõe soluções, mas nenhuma é oficial ainda. Na prática, o que eu faço é injetar tokens via variáveis de ambiente no momento do spawn do server. Funciona, mas é um hack — e não escala para cenários onde o token precisa ser refreshed durante a sessão. Session state MCP sessions são stateless por padrão. Se o server crashar e reiniciar, todo o contexto acumulado se perde. O paper do arxiv propõe um State Checkpoint Pattern, mas ninguém implementou isso nos SDKs oficiais ainda. Se seu agente depende de estado acumulado ao longo de uma conversa (e a maioria depende), você precisa implementar persistência por conta própria. Streaming O suporte a streaming de respostas longas é inconsistente entre implementações. O SDK TypeScript lida bem com SSE, mas o SDK Python tem edge cases com backpressure que podem causar memory leaks em sessões longas. Já perdi horas debugando isso. Observabilidade Não existe um padrão de tracing entre client e servers MCP. Se uma cadeia de 5 tool calls falha, boa sorte descobrindo onde foi. Eu adaptei OpenTelemetry manualmente nos meus servers, mas deveria ser built-in. Roadmap 2026: o que vem por aí O The New Stack publicou o roadmap que a comunidade está trabalhando. Os pontos mais relevantes:Auth padronizado: OAuth 2.1 como padrão de autenticação para MCP servers. Finalmente. Streamable HTTP transport: substituição do SSE por um transport mais robusto para produção. Registry protocol: um padrão para discovery de MCP servers — tipo um DNS para ferramentas de agentes. Elicitation: capacidade do server pedir informação adicional ao usuário via o client, sem interromper o fluxo do agente.Se o auth padronizado e o registry saírem no Q2 como prometido, MCP vira um protocolo enterprise-ready de verdade. Até lá, prepare-se para escrever bastante glue code. Veredito MCP não é mais experimental. 97 milhões de downloads mensais e suporte universal dos providers transformaram o protocolo em padrão de facto. O paper arxiv 2603.13417 é leitura obrigatória para quem está deployando agentes — os design patterns economizam semanas de tentativa e erro. Mas "padrão de facto" não significa "maduro". Auth, state e observabilidade são problemas reais que você vai enfrentar. O roadmap promete resolver boa parte disso em 2026, e a velocidade da comunidade (de 2M para 97M downloads em 16 meses) me dá alguma confiança. Se você ainda não tem MCP servers no seu stack de agentes, comece com o Gateway Pattern e um server simples como o exemplo acima. Mantenha o escopo pequeno, instrumente tudo, e prepare-se para reescrever o auth quando o padrão OAuth 2.1 sair. Paper: arxiv.org/abs/2603.13417. SDK TypeScript: @modelcontextprotocol/sdk. SDK Python: mcp no PyPI. Vai lá, monta um server, quebra em produção, e me conta o que deu errado.