Sistemas
Multiagentes.

Quando Um Agente Não Basta — e Quando Multiagente Não é a Resposta

Assim como uma equipe de software separa arquitetos, desenvolvedores e QA em vez de sobrecarregar uma única pessoa, sistemas complexos se beneficiam de especialistas. Mas quando dividir em múltiplos agentes realmente ajuda — e quando só piora a latência?

Os 4 Sinais de Limite de um Agente Único (Figura 4.1)

• Context Overload: um único agente tenta acompanhar tarefas demais e perde o foco.
• Jack of all trades, master of none: gera resultados medíocres em tudo em vez de excelência em domínios específicos.
• No Parallelization: tudo roda em sequência, mesmo tarefas independentes.
• Difficult to Debug: quando algo falha, é difícil isolar qual parte do raciocínio quebrou.

Ao dividir em especialistas, você ganha Specialization, Parallel Processing, Better Debugging e Scalability — apoiados nos 4 pilares de design (pág. 102): Orchestration, Specialization, Collaboration e Coordination.

Conexão com o livro · Capítulo 4 (págs. 99–102 e 126) · When one agent isn't enough e When multi-agent is not the answer.

Padrão Supervisor-Worker e Composabilidade com Agent as a Tool

Se três agentes especialistas rodam em paralelo e devolvem três blocos desconexos de fatos, quem sintetiza a resposta final para o usuário? Conheça o padrão mais comum em produção e como torná-lo reutilizável.

Supervisor-Worker (Figura 4.2) e Trade-offs de Produção

No padrão Supervisor-Worker, um agente coordenador recebe a pergunta (ex.: “Should I invest in electric vehicle companies?”), delega em paralelo para news_agent, financial_agent e sentiment_agent (os workers nunca falam diretamente entre si) e sintetiza tudo em uma recomendação coerente. Trade-offs (pág. 106): o supervisor pode virar gargalo e ponto único de falha (SPOF) — mitigados em produção com escalonamento horizontal, caching, prompt caching, health checks e failover com checkpoints.

Agent as Tool (Figuras 4.3 a 4.5) e Fluxo de Dados (Figura 4.6)

Enquanto no Supervisor-Worker clássico os workers são acoplados àquele supervisor, o padrão Agent as Tool encapsula um agente especialista com nome e descrição clara para que qualquer orquestrador o descubra e chame como uma ferramenta comum (ex.: math_tutor reutilizado por um teacher, por um tutoring_assistant e por um homework_grader, podendo ele próprio usar sub-agentes como calculator_agent). Na Figura 4.6, o fluxo completo passa por: User Input → Router Agent → Specialist Agents (em paralelo) → Aggregator → Output.

from strands import Agent

news_agent = Agent(
    name="news_agent",
    system_prompt="Research recent news about companies and industries.",
    tools=[news_search_tool]
)
financial_agent = Agent(
    name="financial_agent",
    system_prompt="Analyze financial metrics, stock performance, and valuations.",
    tools=[financial_api_tool]
)
sentiment_agent = Agent(
    name="sentiment_agent",
    system_prompt="Analyze market sentiment from social media and forums.",
    tools=[sentiment_analysis_tool]
)

research_supervisor = Agent(
    name="research_supervisor",
    system_prompt="""You are a market research coordinator.
    1. Analyze the question to determine what information is needed.
    2. Delegate to news_agent, financial_agent, and sentiment_agent.
    3. Synthesize their responses into a coherent investment recommendation.""",
    tools=[news_agent.as_tool(), financial_agent.as_tool(), sentiment_agent.as_tool()]
)

print(research_supervisor("Should I invest in electric vehicle companies?"))

Conexão com o livro · Capítulo 4 (págs. 102–112) · Supervisor-worker, Agent as a tool e Data flow (Figuras 4.2 a 4.6).

Técnicas de Comunicação: Sequencial, Paralela (Votação) e Colaborativa

Agentes isolados são apenas scripts separados. O que os transforma em um sistema inteligente é como trocam informação: em fila estrita, em paralelo simultâneo ou negociando trade-offs em múltiplas rodadas.

Conexão com o livro · Capítulo 4 (págs. 112–116) · Sequential, Parallel (Voting) e Collaborative communication.

Orquestração Avançada: Swarm (Handoffs) vs. Graph Pattern (DAG)

Quando você não consegue prever o caminho exato antes da execução, use Swarm. Quando a auditoria e o compliance exigem que a decisão siga trilhos explícitos no código, use Graph (DAG).

Swarm vs. Graph (Tabela 4.1)

from strands import Agent
from strands.multiagent import Swarm, GraphBuilder

# 1. SWARM: Handoffs autônomos com proteção contra loops infinitos ("ping-pong")
swarm = Swarm(
    [level_designer, game_designer, narrative_designer, difficulty_balancer],
    entry_point=level_designer,
    max_handoffs=20,
    max_iterations=20,
    repetitive_handoff_detection_window=8,
    repetitive_handoff_min_unique_agents=3
)

# 2. GRAPH (DAG): Roteamento explícito e auditável para pedidos (Figura 4.9)
builder = GraphBuilder()
builder.add_node(risk_analyzer, "risk_analysis")
builder.add_node(manual_review_agent, "manual_review")
builder.add_node(fraud_detection_agent, "fraud_check")
builder.add_node(inventory_agent, "inventory_check")
builder.add_node(shipping_agent, "shipping")
builder.add_node(refund_agent, "refund")

builder.add_edge("risk_analysis", "manual_review", condition=is_high_value)
builder.add_edge("risk_analysis", "fraud_check", condition=is_suspicious)
builder.add_edge("risk_analysis", "inventory_check", condition=is_low_risk)
builder.add_edge("manual_review", "inventory_check")
builder.add_edge("fraud_check", "inventory_check")
builder.add_edge("inventory_check", "shipping", condition=is_in_stock)
builder.add_edge("inventory_check", "refund", condition=is_out_of_stock)
builder.set_entry_point("risk_analysis")

graph = builder.build()
result = graph("Order #12345: $1,500 laptop from new customer")

Conexão com o livro · Capítulo 4 (págs. 117–125) · Swarm patterns e Graph pattern (Figura 4.9 e Tabela 4.1).

Model Context Protocol (MCP): Arquitetura, Recursos e Simulador N+M

Sem um padrão universal, conectar N modelos/agentes a M ferramentas exige N × M integrações sob medida. Como o MCP reduz isso para N + M com descoberta dinâmica?

Host vs. Client vs. Server, Transportes e Ciclo de Vida (Figuras 5.1 a 5.12)

• Host vs. MCP Client (Figura 5.5): o Host é a aplicação (IDE, chat, agente) que gerencia UI, memória e prompts; dentro dele existem múltiplos MCP Clients, cada um mantendo conexão isolada 1:1 com um único MCP Server.
• Transportes e JSON-RPC: mensagens em JSON-RPC UTF-8 trafegando via Stdio (local/CLI), SSE / Streamable HTTP (remoto/streaming com especificação de autorização HTTP) ou transportes customizados (ex.: gRPC). Ciclo de vida em 3 fases: Initialization → Operation → Shutdown.
• Server Features (Figura 5.6): Tools (ações executáveis: tools/list, tools/call), Resources (dados somente leitura com URI templates ex. travel://activities/{city}/{category}: resources/list, resources/read) e Prompt Templates (instruções parametrizadas centralizadas).
• Client Features (Figuras 5.11–5.12): Sampling (servidor pede ao cliente para rodar análise de IA com aprovação do usuário), Roots (limites de diretórios no sistema de arquivos) e Elicitation (modos form e URL para pedir dados faltantes em tempo de execução).

Conexão com o livro · Capítulo 5 (págs. 130–147) · Model Context Protocol (Figuras 5.1 a 5.12).

Construindo Servidores e Clientes MCP na Prática com FastMCP e Strands

Como consumir servidores MCP prontos da AWS em duas linhas e como escrever seu próprio servidor (server.py) e cliente (client.py) em Python usando FastMCP?

Servidores Prontos via uvx e FastMCP vs. FastAPI (págs. 147–154)

Você pode conectar um agente Strands diretamente a servidores prontos como o AWS Documentation MCP Server (awslabs.aws-documentation-mcp-server@latest) usando MCPClient + stdio_client. Quando precisar construir o seu em Python: escolha FastMCP para servidores MCP dedicados com menos boilerplate (@mcp.tool(), @mcp.resource(), @mcp.prompt()), ou FastAPI quando o MCP for apenas parte de uma API REST corporativa maior.

# --- server.py (FastMCP) ---
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("WeatherServer")

@mcp.tool()
def get_weather(city: str) -> str:
    """Get the current weather for a given city."""
    fake_weather = {"london": "15°C, cloudy", "paris": "18°C, sunny", "tokyo": "22°C, humid"}
    result = fake_weather.get(city.lower())
    return f"Weather in {city}: {result}" if result else f"No weather data for '{city}'."

@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Return a personalized greeting."""
    return f"Hello, {name}! Welcome to the WeatherServer."

@mcp.prompt()
def weather_report(city: str) -> str:
    """Generate a prompt asking for a weather report."""
    return f"Please provide a detailed weather report for {city}, including temperature, humidity, and forecast."

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

# --- client.py (MCP Client) ---
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(command="python", args=["server.py"])
    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            print(await session.call_tool("get_weather", {"city": "London"}))
            print(await session.read_resource("greeting://Developer"))
            print(await session.get_prompt("weather_report", {"city": "Tokyo"}))

if __name__ == "__main__":
    asyncio.run(main())

Conexão com o livro · Capítulo 5 (págs. 147–154) · Building and using MCP servers, FastMCP vs FastAPI, server.py e client.py.

Agent2Agent (A2A) Protocol: Agent Cards, Bindings e Estilos de Execução

Reduzir um agente complexo a uma simples chamada de ferramenta MCP limita sua autonomia multi-turno. Como permitir que agentes opacos construídos em LangGraph, CrewAI, Google ADK ou Strands colaborem como pares?

Arquitetura, Descoberta e os 4 Modos de Comunicação A2A (Figuras 5.14 a 5.22)

Originalmente lançado pelo Google, fundido ao ACP da IBM e mantido pela Linux Foundation, o A2A mantém agentes remotos opacos (sem expor código, prompts ou banco interno). Cada servidor publica um Agent Card em /.well-known/agent-card.json declarando capabilities, skills e url. A comunicação segue o modelo Message → Task (rastreável via GetTask) → Artifact (contendo Parts) sobre bindings HTTP, gRPC ou JSON-RPC, em quatro estilos:

Na implementação, você escolhe entre 3 camadas: A2A SDK (a2a-sdk, primitivas de baixo nível), Strands A2A (strands-agents[a2a] com A2AServer e A2AAgent) ou plataformas gerenciadas como LangSmith Agent Server e Amazon Bedrock AgentCore Runtime.

Conexão com o livro · Capítulo 5 (págs. 155–167) · Agent2Agent protocol (Figuras 5.14 a 5.22).

Laboratório Completo: Sistema Multiagente de Viagens com Strands A2A + MCP

Vamos juntar os dois protocolos em um sistema real (Figura 5.23): um Travel Orchestrator (cliente A2A) descobre e consulta o Weather Agent na porta 9001 e o Flights Agent na porta 9002.

Os 3 Serviços do Projeto a2aexample/ (págs. 168–174)

# --- weather_agent_server.py (Porta 9001) ---
from a2a.types import AgentSkill
from strands import Agent
from strands.multiagent.a2a import A2AServer

WEATHER_SKILLS = [
    AgentSkill(
        id="get_weather", name="Get Weather",
        description="Returns current weather conditions for a given city.",
        tags=["weather", "temperature", "forecast"],
        examples=["What is the weather in London?"]
    )
]
weather_agent = Agent(name="Weather Agent", tools=[get_weather], callback_handler=None)

if __name__ == "__main__":
    server = A2AServer(
        agent=weather_agent, host="127.0.0.1", port=9001,
        skills=WEATHER_SKILLS, version="1.0.0", enable_a2a_compliant_streaming=True
    )
    server.serve()

# --- flights_agent_server.py segue o mesmo padrão na porta 9002 ---

# --- travel_orchestrator.py (Cliente A2A) ---
import asyncio, httpx
from uuid import uuid4
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, TextPart

async def send_a2a_message(base_url: str, text: str) -> str:
    async with httpx.AsyncClient(timeout=60) as hc:
        card = await A2ACardResolver(httpx_client=hc, base_url=base_url).get_agent_card()
        client = ClientFactory(ClientConfig(httpx_client=hc, streaming=False)).create(card)
        msg = Message(kind="message", role=Role.user, parts=[Part(TextPart(kind="text", text=text))], message_id=uuid4().hex)
        async for event in client.send_message(msg):
            if hasattr(event, "artifacts") and event.artifacts:
                return "".join(p.root.text for a in event.artifacts for p in a.parts)
        return ""

async def main(city="London", origin="New York", destination="London"):
    print("Weather:", await send_a2a_message("http://127.0.0.1:9001", f"What is the weather in {city}?"))
    print("Flights:", await send_a2a_message("http://127.0.0.1:9002", f"Search flights from {origin} to {destination}"))

asyncio.run(main())

Conexão com o livro · Capítulo 5 (págs. 168–175) · Building a multi-agent system with A2A and MCP (Figura 5.23).