Laboratório: Orquestração Multiagente com Claude Agent SDK, OpenAI Agents SDK e MCP em Python

Roteiro para estudo em ambiente de laboratório ou de teste. Execute apenas em equipamentos, redes e contas que você tenha autorização para usar, com backup e homologação próprios. Comandos podem causar indisponibilidade ou perda de dados. Software e marcas citados pertencem aos seus titulares. Termos de Uso.

Duração estimada

120 min

Passos

12

Validações

5

Versões da documentação consultada

Python 3.10 ou superior · Claude Agent SDK a versão estável atual · OpenAI Agents SDK a versão estável atual · Redis a versão estável atual · FastAPI a versão estável atual · Pydantic a versão estável atual

Topologia

Um servidor Python central (orquestrador) recebe tarefas via API REST. Ele distribui para dois agentes: um Claude (via Claude Agent SDK) e um OpenAI (via OpenAI Agents SDK), cada um rodando localmente. As tarefas são enfileiradas em Redis. O orquestrador coleta os resultados, registra logs de tracing e executa testes automáticos. Um script simula revisão humana. Todos os componentes rodam em localhost, usando portas distintas (exemplo: 127.0.0.1:8000 para API, 6379 para Redis).

        [Usuário/API REST]
                |
         [Orquestrador Python]
           /           \
  [Claude Agent]   [OpenAI Agent]
           \           /
            [Fila Redis]
                |
         [Avaliação/Testes]
                |
       [Revisão Humana Simulada]

Equipamentos e software

Roteiro

1. Preparar ambiente Python e instalar dependências

   Isolar o ambiente evita conflitos. Instale SDKs, FastAPI, Redis client e utilitários.

   python3 -m venv multiagentes-lab
   source multiagentes-lab/bin/activate
   pip install fastapi uvicorn redis pydantic
   pip install claude-agent-sdk-python
   pip install openai-agentsCopiar

   O que você deve ver: Ambiente virtual ativado e pacotes instalados sem erro.

2. Instalar e iniciar Redis localmente

   A fila Redis permite desacoplar a orquestração dos agentes. Redis é leve e fácil de rodar local.

   sudo apt-get update && sudo apt-get install redis-server -y
   sudo systemctl start redis-server
   redis-cli pingCopiar

   O que você deve ver: Resposta 'PONG' no terminal.

3. Configurar variáveis de ambiente para segredos

   Nunca exponha chaves no código. Use variáveis para as APIs Claude e OpenAI.

   export CLAUDE_API_KEY='sua-chave-anthropic'
   export OPENAI_API_KEY='sua-chave-openai'Copiar

   O que você deve ver: Variáveis setadas na sessão. Não aparece saída ao exportar.

4. Criar API REST de entrada de tarefas com FastAPI

   A API recebe tarefas e as coloca na fila Redis. Use Pydantic para validação.

   cat > api.py <<'EOF'
   from fastapi import FastAPI
   from pydantic import BaseModel
   import redis
   import json
   
   app = FastAPI()
   r = redis.Redis(host='localhost', port=6379, db=0)
   
   class Tarefa(BaseModel):
       tipo: str
       dados: dict
   
   @app.post('/tarefa')
   def nova_tarefa(tarefa: Tarefa):
       r.lpush('tarefas', json.dumps(tarefa.dict()))
       return {'status': 'enfileirada'}
   EOFCopiar

   O que você deve ver: Arquivo api.py criado. Nenhum erro.

5. Implementar o orquestrador Python

   O orquestrador lê da fila, decide qual agente executar e faz tracing. Separe funções para clareza.

   cat > orquestrador.py <<'EOF'
   import redis, json, logging, os
   from claude_agent_sdk import ClaudeAgent
   from openai_agents import Agent, Runner
   
   logging.basicConfig(filename='tracing.log', level=logging.INFO)
   r = redis.Redis(host='localhost', port=6379, db=0)
   
   # Exemplos de prompts versionados
   PROMPT_CLAUDE = 'v1: Responda como especialista em análise de sentimentos.'
   PROMPT_OPENAI = 'v1: Responda como especialista em resumo de texto.'
   
   while True:
       tarefa_raw = r.rpop('tarefas')
       if not tarefa_raw:
           continue
       tarefa = json.loads(tarefa_raw)
       tipo = tarefa['tipo']
       dados = tarefa['dados']
       if tipo == 'sentimento':
           agent = ClaudeAgent(api_key=os.environ['CLAUDE_API_KEY'], instructions=PROMPT_CLAUDE)
           resposta = agent.run_sync(dados['texto'])
       elif tipo == 'resumo':
           agent = Agent(name='openai-resumidor', instructions=PROMPT_OPENAI)
           resposta = Runner.run_sync(agent, dados['texto'])
       else:
           resposta = 'Tipo de tarefa não suportado.'
       logging.info(f"Tarefa: {tarefa} | Resposta: {resposta}")
       r.lpush('resultados', json.dumps({'tarefa': tarefa, 'resposta': resposta}))
   EOFCopiar

   O que você deve ver: Arquivo orquestrador.py criado. Nenhum erro.

6. Adicionar guardrails de entrada e saída

   Valide tipos e tamanho dos dados antes de enviar ao agente. Rejeite entradas inválidas.

   cat >> orquestrador.py <<'EOF'
   # Após carregar tarefa, antes de rodar agente:
   if 'texto' not in dados or not isinstance(dados['texto'], str) or len(dados['texto']) > 1000:
       resposta = 'Entrada inválida.'
       logging.warning(f"Tarefa inválida: {tarefa}")
       r.lpush('resultados', json.dumps({'tarefa': tarefa, 'resposta': resposta}))
       continue
   EOFCopiar

   O que você deve ver: Arquivo orquestrador.py atualizado com validação. Nenhum erro.

7. Implementar avaliação automática dos resultados

   Teste se a resposta do agente atende ao esperado. Use casos de teste simples.

   cat > avaliador.py <<'EOF'
   import redis, json
   r = redis.Redis(host='localhost', port=6379, db=0)
   # Exemplo de caso de teste
   casos = [
       {'tipo': 'sentimento', 'dados': {'texto': 'O produto é ótimo!'}, 'esperado': 'positivo'},
       {'tipo': 'resumo', 'dados': {'texto': 'Python é uma linguagem de programação.'}, 'esperado': 'linguagem de programação'}
   ]
   for caso in casos:
       r.lpush('tarefas', json.dumps(caso))
   
   print('Aguardando resultados...')
   for _ in casos:
       res_raw = r.brpop('resultados')[1]
       res = json.loads(res_raw)
       print(f"Tarefa: {res['tarefa']}, Resposta: {res['resposta']}")
   EOFCopiar

   O que você deve ver: Arquivo avaliador.py criado. Nenhum erro.

8. Simular revisão humana para ações irreversíveis

   Antes de executar ações críticas, envie para revisão. Aqui, apenas registre e aguarde aprovação manual.

   cat > revisao.py <<'EOF'
   import redis, json
   r = redis.Redis(host='localhost', port=6379, db=0)
   while True:
       res_raw = r.brpop('resultados')[1]
       res = json.loads(res_raw)
       print(f"Revisão humana: {res}")
       # Simule aprovação
       r.lpush('aprovados', json.dumps(res))
   EOFCopiar

   O que você deve ver: Arquivo revisao.py criado. Nenhum erro.

9. Rodar todos os serviços em paralelo

   Execute API, orquestrador, avaliador e revisão em terminais separados. Isso simula produção.

   uvicorn api:app --reload --port 8000
   # Em outro terminal
   python orquestrador.py
   # Em outro terminal
   python avaliador.py
   # Em outro terminal
   python revisao.pyCopiar

   O que você deve ver: Todos os serviços rodando. API responde, orquestrador processa, avaliador testa, revisão imprime resultados.

10. Enviar tarefas de teste via API

    Use curl ou httpie para testar entrada de tarefas.

    curl -X POST 'http://127.0.0.1:8000/tarefa' -H 'Content-Type: application/json' -d '{"tipo": "sentimento", "dados": {"texto": "O serviço foi ruim."}}'Copiar

    O que você deve ver: {"status":"enfileirada"}

11. Verificar logs de tracing

    Confirme que cada passo foi registrado. O tracing é essencial para auditoria e depuração.

    tail -f tracing.logCopiar

    O que você deve ver: Linhas de log detalhando tarefas, respostas e avisos de entrada inválida.

12. Revisar resultados aprovados

    Veja as respostas aprovadas pela revisão humana simulada.

    redis-cli LRANGE aprovados 0 -1Copiar

    O que você deve ver: Lista de respostas em JSON, aprovadas.

Como saber que funcionou

Troubleshooting

Segurança e hardening

• Nunca exponha chaves de API em código ou logs
• Restrinja acesso ao Redis para localhost
• Limite tamanho e tipo dos dados aceitos na API
• Registre todas as ações críticas para auditoria
• Implemente revisão humana para tarefas sensíveis
• Descarte dados pessoais após o uso, conforme LGPD

Checklist final

• Ambiente Python isolado e dependências instaladas
• Redis rodando e acessível
• API FastAPI aceita tarefas
• Orquestrador processa e registra logs
• Avaliação automática compara resultados
• Revisão humana simula aprovação
• Tracing e filas funcionam de ponta a ponta