Pipeline Multi-Agente
para CULTIVA IA

# research_agent.py — A2A Server
from a2a.types import AgentCard, AgentSkill, AgentCapabilities
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.apps.starlette import A2AStarletteApplication
from a2a.types import Message, TextPart, DataPart
from a2a.types import TaskState, TaskStatus
import uvicorn, asyncio

agent_card = AgentCard(
    name="research-agent",
    description="Investiga mercado y competencia para campañas",
    url="https://research.cultivaia.app",
    version="1.0.0",
    capabilities=AgentCapabilities(
        streaming=True,
        pushNotifications=True,
    ),
    skills=[
        AgentSkill(
            id="web-search",
            name="Búsqueda Web",
            description="Investiga tendencias, competidores y público",
            tags=["research", "market", "competitors"],
            examples=["Analiza el mercado de apps de reserva para peluquerías"],
        ),
    ],
    defaultInputModes=["text/plain", "application/json"],
    defaultOutputModes=["application/json"],
)

class ResearchExecutor(AgentExecutor):
    async def execute(self, ctx: RequestContext,
                      eq: EventQueue):
        query = ctx.get_user_message().parts[0].text

        # Emitir estado working con SSE
        await eq.enqueue_event(TaskStatus(
            state=TaskState.working,
            message=Message(role="agent",
                parts=[TextPart(text="🔍 Analizando mercado...")])
        ))

        results = await self._research(query)

        # Emitir resultado como DataPart (JSON)
        await eq.enqueue_event(TaskStatus(
            state=TaskState.completed,
            message=Message(role="agent",
                parts=[DataPart(data=results)])
        ))

    async def _research(self, query: str) -> dict:
        await asyncio.sleep(2)  # LLM call
        return {
            "target_audience": "Dueños peluquería 35-55",
            "pain_points": ["citas por teléfono", "no-shows"],
            "competitors": ["Planfy", "Setmore", "SimplyBook"],
            "differentiators": ["IA predictiva de hueco", "sin comisión"],
        }

    async def cancel(self, ctx: RequestContext,
                    eq: EventQueue):
        await eq.enqueue_event(
            TaskStatus(state=TaskState.canceled))

handler = DefaultRequestHandler(
    agent_executor=ResearchExecutor(),
    task_store=InMemoryTaskStore(),
)
app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=handler,
)
uvicorn.run(app.build(), host="0.0.0.0", port=8001)

# orchestrator.py — Pipeline A2A secuencial
from a2a.client import A2AClient
from a2a.types import MessageSendParams, SendMessageRequest
from a2a.types import Message, TextPart, DataPart
import asyncio, json

async def build_content_pipeline(brief: str):
    """Pipeline: research → write → review"""

    # 1. Descubrir agentes via Agent Cards
    research = await A2AClient.get_client_from_agent_card_url(
        "https://research.cultivaia.app/.well-known/agent.json"
    )
    writer = await A2AClient.get_client_from_agent_card_url(
        "https://writer.cultivaia.app/.well-known/agent.json"
    )
    reviewer = await A2AClient.get_client_from_agent_card_url(
        "https://reviewer.cultivaia.app/.well-known/agent.json"
    )

    # 2. STEP 1 — Investigar mercado
    research_resp = await research.send_message(
        SendMessageRequest(params=MessageSendParams(
            message=Message(role="user",
                parts=[TextPart(text=f"Investiga: {brief}")])
        ))
    )
    market_data = research_resp.status.message.parts[0].data
    print(f"✅ Research completado: {market_data['target_audience']}")

    # 3. STEP 2 — Generar copies (con streaming)
    async for event in writer.send_message_streaming(
        SendMessageRequest(params=MessageSendParams(
            message=Message(role="user",
                parts=[DataPart(data={
                    "brief": brief,
                    "market_data": market_data,
                    "channels": ["web", "email", "instagram"],
                })])
        ))
    ):
        if hasattr(event, "status") and event.status.message:
            part = event.status.message.parts[0]
            if hasattr(part, "text"):
                print(part.text, end="", flush=True)
            elif hasattr(part, "data"):
                content_package = part.data

    # 4. STEP 3 — Revisar con human-in-the-loop
    review_resp = await reviewer.send_message(
        SendMessageRequest(params=MessageSendParams(
            message=Message(role="user",
                parts=[DataPart(data=content_package)])
        ))
    )

    # Manejar input-required (aprobación humana)
    if review_resp.status.state.value == "input-required":
        print("\n⚠️ Revisión manual requerida")
        return review_resp

    return content_package

# Ejecutar pipeline
brief = """Lanzamiento PeluquIA: app de reservas con IA
para peluquerías. Target: dueños 35-55 años.
Tono: cercano, moderno, sin tecnicismos."""

result = asyncio.run(build_content_pipeline(brief))