Technical Exploration Report: Prometheus Frontend + Client-Agent A2A Integration¶

@cl.on_message
async def on_message(message: cl.Message):
    await message_handler(message)

# Line 101-115
payload = {"message": message.content, "user_id": "user_123", "context": {}, "files": []}

response = requests.post(
    f"{settings.platform_url}/api/chat",
    json=payload,
    headers={"Authorization": "Bearer d8Q0ICXp1AUcGH0mLbkG-lgsUX6f_5RY7pu3KRrHUAE"}
)
data = response.json()

{
  "message": "Design a Rankine cycle...",
  "user_id": "user_123",
  "context": {},
  "files": []
}

{
  "type": "workflow",
  "session_id": "uuid",
  "response": "Starting workflow...",
  "workflow_id": "wf_..."
}

{"event_type": "workflow_start", "total_steps": 3}
{"event_type": "step_start", "agent": "Research Agent", "prompt": "...", "step_index": 0, "total_steps": 3}
{"event_type": "step_complete", "response": "...", "artifact_url": "/api/demo/ifc"}
{"event_type": "workflow_complete"}

message = {
    "jsonrpc": "2.0",
    "id": request_id,  # UUID for request tracking
    "method": "message/send",
    "params": {
        "message": {
            "messageId": message_id,
            "role": "user",
            "parts": [
                {
                    "type": "text",
                    "text": json.dumps(kwargs)  # Parameters as JSON string in message text
                }
            ]
        },
        "configuration": {
            "blocking": True,  # Wait for completion before returning
        },
        "context": {
            "skillId": skill_id,
            "conversationId": kwargs.get("conversation_id", str(uuid.uuid4()))
        }
    }
}

class ConversationResponse(BaseModel):
    conversation_id: str
    status: str  # "needs_info" | "plan_ready" | "executing" | "completed" | "recommendation" | "answer"
    message: Optional[str] = None

    # When status == "needs_info"
    followup_questions: Optional[List[Dict]] = None

    # When status == "plan_ready"
    plan: Optional[Dict] = None

    # When status == "completed"
    report: Optional[str] = None
    tool_results: Optional[Dict] = None

    # When status == "recommendation" or "answer"
    response_text: Optional[str] = None
    recommended_materials: Optional[List[str]] = None
    artifacts: Optional[List[Dict[str, Any]]] = None
    message_metadata: Optional[Dict[str, Any]] = None

class AgentState(TypedDict):
    # Multi-turn tracking
    conversation_id: Optional[str]
    user_query: Optional[str]
    conversation_history: List[Dict]  # Full history across turns

    # Routing & planning
    routing_decision: Optional[Dict[str, Any]]
    plan: Optional[Dict]
    missing_info: List[str]
    awaiting_approval: bool

    # Query parameters
    material_id: Optional[str]
    analysis_types: Optional[List[str]]
    context: Optional[Dict[str, Any]]
    question: Optional[str]

    # Memory & artifacts
    retrieved_memories: Optional[List[Dict[str, Any]]]
    extracted_factoids: Optional[List[str]]
    num_memories_retrieved: Optional[int]
    num_memories_stored: Optional[int]

    # Results
    tool_results: Dict[str, Any]
    final_report: Optional[str]

# timeout_seconds = 900.0  # 15 minutes
async with httpx.AsyncClient(timeout=timeout_seconds) as client:
    response = await client.post(agent_url, json=message, headers=headers)
    result = response.json()

class ConversationRequest(BaseModel):
    query: Optional[str] = ""                    # Natural language query
    conversation_id: Optional[str] = None        # Multi-turn identifier
    followup_answers: Optional[Dict] = None      # Answers to prior questions
    plan_approved: Optional[bool] = False        # Approval gate
    artifacts: Optional[List[Dict[str, Any]]] = None  # Artifact catalog
    workflow_id: Optional[str] = None            # Explicit workflow selection
    workflow_version: Optional[str] = None
    selection_source: Optional[str] = None       # "catalog|matcher|dynamic_router"

class PrometheusAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 1. Extract parameters from message
        params = self._extract_params(context)

        # 2. Send "working" status
        await self._emit_status(
            context, event_queue,
            state=TaskState.working, final=False
        )

        # 3. Call agent-specific _execute() implementation
        result = await self._execute(params, context, tracked_queue)

        # 4. Send result artifact (if not already sent)
        if not tracked_queue.seen_artifact:
            await self._emit_json_artifact(context, event_queue, payload=result)

        # 5. Send completion status
        await self._emit_status(
            context, event_queue,
            state=TaskState.completed, final=True
        )

async def _emit_status(
    self,
    context: RequestContext,
    event_queue: EventQueue,
    state: TaskState,  # TaskState.working | completed | failed
    final: bool,
    message: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> None:
    status_message = None
    if message:
        status_message = Message(
            messageId=str(uuid.uuid4()),
            role="agent",
            parts=[Part(root=TextPart(text=message))],
            contextId=context.context_id,
            taskId=context.task_id,
        )

    await event_queue.enqueue_event(
        TaskStatusUpdateEvent(
            taskId=context.task_id,
            contextId=context.context_id,
            final=final,
            status=TaskStatus(state=state, message=status_message),
            metadata=metadata,
        )
    )

@dataclass
class ArtifactReference:
    artifact_id: str
    conversation_id: str
    name: str
    description: str
    uri: str  # Location of actual file (filesystem, HTTP, DeepLynx, etc.)
    mime_type: Optional[str] = None
    size_bytes: Optional[int] = None
    producer_agent_id: Optional[str] = None
    created_at: str = field(default_factory=_utc_now_iso)
    tags: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)

# Auto-registration in send_json_artifact
parsed_ref = self._payload_to_artifact_reference(
    payload,
    fallback_name=name,
    conversation_id=context.context_id,
    producer_agent_id=self.agent_name.lower()
)
if parsed_ref is not None:
    registered = self.register_artifact_reference(context, parsed_ref)

await agent.send_status_update(
    context, event_queue,
    state=TaskState.working,
    message="Processing material data...",
    final=False
)

artifact_id = await agent.send_json_artifact(
    context, event_queue,
    payload={"analysis": result},
    name="analysis_result"
)

# Or for external files:
await agent.send_artifact_reference(
    context, event_queue,
    name="Generated CAD Model",
    description="3D mesh in OBJ format",
    uri="file:///tmp/mesh.obj",
    mime_type="model/obj"
)

# Current (message_handler.py:104)
response = requests.post(
    f"{settings.platform_url}/api/chat",
    json={"message": message.content, "user_id": "user_123", ...},
    headers={"Authorization": "Bearer ..."}
)

// Proposed (will be in new message_handler or chat utility)
const response = await fetch('http://localhost:8000', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer ' + agentToken
  },
  body: JSON.stringify({
    jsonrpc: "2.0",
    id: requestId,
    method: "message/send",
    params: {
      message: {
        messageId: uuid(),
        role: "user",
        parts: [{
          type: "text",
          text: JSON.stringify({
            query: messageContent,
            conversation_id: convId,
            artifacts: artifactList  // Optional: pass artifact metadata
          })
        }]
      },
      configuration: { blocking: true },
      context: {
        skillId: "analyze_material",  // Or "coordinate_analysis"
        conversationId: convId
      }
    }
  })
})

# Need to add in message_handler.py:
conversation_id = None  # Persist this across turns
artifacts_catalog = []  # Track available artifacts

async def message_handler(message):
    global conversation_id, artifacts_catalog

    if conversation_id is None:
        conversation_id = str(uuid.uuid4())

    # Build request
    request = ConversationRequest(
        query=message.content,
        conversation_id=conversation_id,
        artifacts=artifacts_catalog
    )

    # Call A2A agent
    response = await call_a2a_agent(
        agent_url="http://localhost:8000",
        request=request
    )

    # Extract artifacts for next turn
    if response.artifacts:
        artifacts_catalog = response.artifacts

# config.yml - add new section
a2a-agent:
  url: "http://client-agent:8000"  # or env var AGENT_URL
  auth_token: "${AGENT_SECRET}"     # env var for Bearer token

# In Chainlit session metadata
@cl.on_chat_start
async def on_chat_start():
    cl.user_session.set("conversation_id", str(uuid.uuid4()))
    # Load prior artifacts if session resumed
    cl.user_session.set("artifacts", [])

# In message_handler
async def message_handler(message):
    conv_id = cl.user_session.get("conversation_id")
    artifacts = cl.user_session.get("artifacts", [])

    # Make A2A call
    response = await call_a2a_client.send(
        query=message.content,
        conversation_id=conv_id,
        artifacts=artifacts
    )

    # Update session
    if response.artifacts:
        cl.user_session.set("artifacts", response.artifacts)

# In message_handler - handle artifacts in response
if response.artifacts:
    artifact_catalog = response.artifacts

    for artifact_meta in artifact_catalog:
        # Show available artifacts in UI
        # (artifact_meta = {artifact_id, name, uri, mime_type, ...})
        await cl.Message(
            content=f"Available: {artifact_meta['name']}",
            elements=[  # Optional: embed viewer if mime_type is known
                # defer to specialized handler
            ]
        ).send()

# New response states to handle
if response.status == "needs_info":
    # Ask follow-up questions
    for q in response.followup_questions:
        action = await cl.AskActionMessage(
            content=q.get("question"),
            actions=[
                cl.Action(name="yes", label="Yes"),
                cl.Action(name="no", label="No"),
            ]
        ).send()

        followup_answers[q.get("id")] = action.get("name")

    # Re-invoke with answers
    response = await call_a2a_client.send(
        query=message.content,
        conversation_id=conv_id,
        followup_answers=followup_answers
    )

if response.status == "plan_ready":
    # Show plan, ask for approval
    plan_text = format_plan(response.plan)
    approval = await cl.AskActionMessage(
        content=plan_text,
        actions=[
            cl.Action(name="approve", label="Proceed"),
            cl.Action(name="reject", label="Modify Plan"),
        ]
    ).send()

    response = await call_a2a_client.send(
        query=message.content,
        conversation_id=conv_id,
        plan_approved=(approval.get("name") == "approve")
    )

// Keep exactly as-is; no protocol changes needed:
- DynamicForm.jsx (form rendering)
- ExodusViewer.jsx (VTK.js 3D visualization)
- IFCViewer.jsx (ThatOpen/OBC 3D visualization)
- MermaidViewer.jsx (diagram rendering)
- FollowUpSuggestions.jsx (quick action buttons)

# Keep Chainlit's built-in auth:
@cl.password_auth_callback
def auth_callback(username: str, password: str):
    for user in settings.local_users:
        if user.get("username") == username and user.get("password") == password:
            return cl.User(...)
    return None

# Keep special handlers for Exodus, IFC, PDF uploads
if element.name.endswith((".exo", ".e")):
    dataset = await process_exodus_upload(Path(element.path))
    viewer = dataset.to_chainlit_element()

# Keep config.yml structure, just add a2a-agent section:
platform-url: "http://localhost:8001"  # (deprecated, remove)
a2a-agent:
  url: "http://localhost:8000"
  auth_token: "${AGENT_SECRET}"

# DELETE: prometheus_platform/backend/
#   - main.py
#   - routers/chat.py
#   - routers/demo_workflow.py
#   - routers/demo.py
#   - orchestrator/
#   - models/
#   - etc.

# DELETE from config.yml:
model: "gpt-oss-120B"
openai_api_key: "..."

# DELETE:
# /api/demo/workflow/run (streaming workflow)
# /api/demo/workflow/status
# /api/demo/ifc
# /api/demo/ifc-download
# /api/demo/exodus
# /api/demo/autodone

# DELETE:
# /api/workflows/{id}/status
# /api/chat/status

# DELETE from message_handler.py:104-115
response = requests.post(
    f"{settings.platform_url}/api/chat",  # ← this backend call
    ...
)

# config.yaml for litellm proxy
model_list:
  - model_name: "material-analyzer"
    litellm_params:
      model: "anthropic/claude-opus-4-6"
      api_base: "http://material-agent:8000"
      api_key: "${AGENT_SECRET}"

  - model_name: "workflow-orchestrator"
    litellm_params:
      model: "openai/gpt-4-turbo"
      api_base: "http://client-agent:8000"
      api_key: "${AGENT_SECRET}"

# prometheus_platform/registry/app.py (simplified)
@app.post("/agents/register")
async def register_agent(registration: AgentRegistration):
    """Register agent and sync to litellm config."""
    registered_agents[registration.agent_id] = registration.dict()

    # Update litellm proxy config
    model_entry = {
        "model_name": registration.agent_id,
        "litellm_params": {
            "model": f"a2a/{registration.agent_id}",  # Custom model identifier
            "api_base": registration.url,
            "api_key": os.getenv(f"AGENT_SECRET_{registration.agent_id}"),
        }
    }
    # Write to litellm config (or call litellm management API if available)

    return {"status": "registered"}

@app.get("/capabilities")
async def list_capabilities():
    """Return flattened skills with metadata for LLM-based routing."""
    capabilities = []
    for agent in registered_agents.values():
        for skill in agent.get("agent_card", {}).get("skills", []):
            capabilities.append({
                "skill_id": skill["id"],
                "agent_id": agent["agent_id"],
                "agent_url": agent["url"],
                "input_schema": skill.get("inputSchema", {}),
                "requires_approval": skill.get("requires_approval", False),
                "expected_duration_seconds": skill.get("expected_duration_seconds"),
                # ... other metadata
            })
    return {"capabilities": capabilities}

# prometheus_platform/frontend/config.yml
a2a-agent:
  url: "http://localhost:8000"  # Client-Agent endpoint
  skill_id: "analyze_material"  # Default skill to invoke

# docker-compose.yml
services:
  client-agent:
    image: prometheus/client-agent:latest
    ports:
      - "8000:8000"
    environment:
      - AGENT_PORT=8000
      - REGISTRY_URL=http://litellm:4000
      - MODEL_NAME=claude-opus-4-6
      - AGENT_SECRET=${AGENT_SECRET}  # Shared secret

  frontend:
    image: prometheus/frontend:latest
    ports:
      - "8500:8500"
    environment:
      - AGENT_URL=http://client-agent:4000
      - AGENT_SECRET=${AGENT_SECRET}

# No changes needed; both use same mechanism:
# 1. Token stored in env var: AGENT_SECRET
# 2. Passed as: Authorization: Bearer ${AGENT_SECRET}
# 3. Validated by A2A framework middleware: BearerTokenMiddleware

# From base.py middleware - already implemented
class BearerTokenMiddleware:
    def __call__(self, request: Request) -> Optional[Dict[str, Any]]:
        auth_header = request.headers.get("Authorization", "")
        if not auth_header.startswith("Bearer "):
            return None
        token = auth_header[7:]  # Strip "Bearer " prefix
        expected_token = os.getenv("AGENT_SECRET", "").strip()
        if token == expected_token:
            return {"authenticated": True}
        return None

# prometheus_platform/frontend/utils/a2a_client.py (new file)
import httpx
import json
import uuid
from typing import Dict, Any, Optional

async def call_agent(
    agent_url: str,
    query: str,
    conversation_id: str,
    agent_secret: str,
    artifacts: Optional[List[Dict]] = None,
    followup_answers: Optional[Dict] = None,
    plan_approved: Optional[bool] = False,
) -> Dict[str, Any]:
    """Call client-agent via A2A protocol."""

    request_id = str(uuid.uuid4())
    message_id = str(uuid.uuid4())

    # Build A2A request
    message = {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": "message/send",
        "params": {
            "message": {
                "messageId": message_id,
                "role": "user",
                "parts": [{
                    "type": "text",
                    "text": json.dumps({
                        "query": query,
                        "conversation_id": conversation_id,
                        "artifacts": artifacts or [],
                        "followup_answers": followup_answers,
                        "plan_approved": plan_approved,
                    })
                }]
            },
            "configuration": {"blocking": True},
            "context": {
                "skillId": "analyze_material",
                "conversationId": conversation_id,
            }
        }
    }

    # Call agent with Bearer token
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {agent_secret}",
    }

    async with httpx.AsyncClient(timeout=900.0) as client:
        response = await client.post(agent_url, json=message, headers=headers)
        response.raise_for_status()
        result = response.json()

    # Extract ConversationResponse from A2A result
    a2a_result = result.get("result", {})

    # Parse response (logic from tool_discovery.py:261-373)
    # Check status.message, then artifacts, then history
    # ... (same extraction logic)

    return parse_conversation_response(a2a_result)

# prometheus_platform/frontend/components/message_handler.py (updated)
import asyncio
from utils.a2a_client import call_agent
from utils.settings import settings

@cl.on_message
async def on_message(message: cl.Message):
    # File handling (unchanged)
    if message.elements:
        # ... existing file upload logic ...
        pass

    # Get or create conversation ID
    conversation_id = cl.user_session.get("conversation_id")
    if not conversation_id:
        conversation_id = str(uuid.uuid4())
        cl.user_session.set("conversation_id", conversation_id)

    # Get artifact catalog from session
    artifacts_catalog = cl.user_session.get("artifacts", [])

    try:
        # Call A2A agent
        response = await call_agent(
            agent_url=settings.agent_url,  # "http://localhost:8000"
            query=message.content,
            conversation_id=conversation_id,
            agent_secret=settings.agent_secret,  # From env var
            artifacts=artifacts_catalog,
        )

        # Handle response based on status
        if response.status == "needs_info":
            # Show follow-up questions, recursively call with answers
            pass
        elif response.status == "plan_ready":
            # Show plan, ask for approval, recursively call with approval
            pass
        elif response.status == "completed" or response.status == "answer":
            # Display response + artifacts
            await cl.Message(content=response.response_text).send()

            # Update artifact catalog for next turn
            if response.artifacts:
                artifacts_catalog = response.artifacts
                cl.user_session.set("artifacts", artifacts_catalog)

    except Exception as e:
        await cl.Message(content=f"Error: {str(e)}").send()

prometheus_platform/
├── frontend/
│   ├── components/message_handler.py           (MAJOR: replace backend calls)
│   ├── utils/a2a_client.py                     (NEW: A2A protocol wrapper)
│   ├── config.yml                              (MINOR: update settings)
│   ├── main.py                                 (MINOR: conversation ID init)
│   └── public/elements/*.jsx                   (NO CHANGES)
│
├── backend/                                     (DELETE ENTIRE FOLDER)
│
├── docs/
│   └── exploration_report.md                   (THIS FILE)
│
└── docs/
    └── A2A_INTEGRATION_GUIDE.md               (NEW: implementation guide)

multi_agent_testbed/
├── agents/client-agent/
│   └── agent.py                                (MINOR: already has ConversationRequest/Response)
│
├── registry/
│   └── app.py                                  (NO CHANGES: keep as-is)
│
└── prometheus_api/
    └── prometheus_api/agent/
        ├── base.py                             (NO CHANGES: already complete)
        └── artifacts.py                        (NO CHANGES: already complete)