Initial commit

Author: staverm

.github/workflows/publish.yml

@@ -0,0 +1,52 @@
+name: Publish Docker Image
+
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - 'v*'
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Log in to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata for Docker
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ghcr.io/${{ github.repository }}
+          tags: |
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}
+            type=raw,value=latest,enable={{is_default_branch}}
+
+      - name: Build and push Docker image
+        id: build
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: Dockerfile
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          platforms: linux/amd64
+
+      - name: Output image digest
+        run: |
+          echo "## Docker Image Published" >> $GITHUB_STEP_SUMMARY
+          echo "**Digest:** \`${{ steps.build.outputs.digest }}\`" >> $GITHUB_STEP_SUMMARY

.gitignore

@@ -0,0 +1,3 @@
+.DS_Store
+__pycache__/
+*.pyc

Dockerfile

@@ -0,0 +1,15 @@
+FROM ghcr.io/astral-sh/uv:python3.13-bookworm
+
+ENV PYTHONUNBUFFERED=1
+
+RUN adduser assessment && mkdir -p /output && chown assessment:assessment /output
+USER assessment
+WORKDIR /home/assessment
+
+COPY pyproject.toml uv.lock ./
+COPY src src
+
+RUN uv sync --locked
+
+ENTRYPOINT ["uv", "run", "python", "src/main.py"]
+EXPOSE 8080

README.md

@@ -0,0 +1,14 @@
+# AgentBeats Assessment Service
+
+Initiates an assessment by sending participants and assessment configuration to the Green agent via A2A. Parses the Green agent's response and serves results as JSON.
+
+The endpoint at `/` returns JSON with a `status` field set to `running` while the assessment is in progress. Once the Green agent finishes, the endpoint returns `status` set to the final A2A task state (e.g. `completed`, `failed`) and `results` containing the parsed artifact data.
+
+## Configuration
+
+| Source | Name | Description |
+|---|---|---|
+| Env | `GREEN_URL` | Base URL of the Green agent (required) |
+| Env | `PARTICIPANTS` | JSON mapping role to participant URL |
+| Env | `ASSESSMENT_CONFIG` | JSON assessment configuration |
+| Arg | `--port` | HTTP server port |

amber-manifest.json5

@@ -0,0 +1,32 @@
+{
+  manifest_version: "0.1.0",
+  program: {
+    image: "ghcr.io/rdi-foundation/agentbeats-assessment:latest",
+    entrypoint: "uv run --no-sync python src/main.py --port 8080",
+    env: {
+      GREEN_URL: "${slots.green.url}",
+      PARTICIPANTS: "${config.participants}",
+      ASSESSMENT_CONFIG: "${config.assessment_config}",
+    },
+    network: {
+      endpoints: [
+        { name: "results", port: 8080 },
+      ],
+    },
+  },
+  config_schema: {
+    type: "object",
+    properties: {
+      participants: { type: "object" },
+      assessment_config: { type: "object" },
+    },
+    required: ["participants", "assessment_config"],
+  },
+  slots: {
+    green: { kind: "a2a" },
+  },
+  provides: {
+    results: { kind: "http", endpoint: "results" },
+  },
+  exports: { results: "results" },
+}

pyproject.toml

@@ -0,0 +1,12 @@
+[project]
+name = "agentbeats-assessment"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "a2a-sdk>=0.3.22",
+    "httpx>=0.28.1",
+    "starlette>=0.52.1",
+    "uvicorn>=0.40.0",
+]

src/assessment.py

@@ -0,0 +1,134 @@
+import json
+from typing import Any
+from uuid import uuid4
+
+import httpx
+from a2a.client import (
+    A2ACardResolver,
+    ClientConfig,
+    ClientFactory,
+)
+from a2a.types import (
+    Artifact,
+    DataPart,
+    Message,
+    Part,
+    Role,
+    TaskArtifactUpdateEvent,
+    TaskStatusUpdateEvent,
+    TextPart,
+)
+
+
+DEFAULT_TIMEOUT = 300
+
+
+def print_parts(parts, task_state: str | None = None):
+    text_parts, data_parts = parse_parts(parts)
+
+    output = []
+    if task_state:
+        output.append(f"[Status: {task_state}]")
+    if text_parts:
+        output.append("\n".join(text_parts))
+    if data_parts:
+        output.extend(json.dumps(item, indent=2) for item in data_parts)
+
+    print("\n".join(output) + "\n")
+
+
+async def send_message(
+    message: str, base_url: str
+) -> tuple[str | None, list[Artifact] | None]:
+    async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as httpx_client:
+        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=base_url)
+        agent_card = await resolver.get_agent_card()
+        config = ClientConfig(
+            httpx_client=httpx_client,
+            streaming=True,
+        )
+        factory = ClientFactory(config)
+        agent_card.url = base_url
+        client = factory.create(agent_card)
+        outbound_msg = Message(
+            kind="message",
+            role=Role.user,
+            parts=[Part(root=TextPart(kind="text", text=message))],
+            message_id=uuid4().hex,
+            context_id=None,
+        )
+
+        artifacts: list[Artifact] | None = None
+        final_status: str | None = None
+
+        async for event in client.send_message(outbound_msg):
+            match event:
+                case Message() as msg:
+                    print_parts(msg.parts)
+
+                case (task, TaskStatusUpdateEvent() as status_event):
+                    status = status_event.status
+                    parts = status.message.parts if status.message else []
+                    print_parts(parts, status.state.value)
+                    final_status = status.state.value
+                    if status.state.value == "completed":
+                        print(task.artifacts)
+                        artifacts = task.artifacts
+
+                case (task, TaskArtifactUpdateEvent() as artifact_event):
+                    print_parts(artifact_event.artifact.parts, "Artifact update")
+
+                case task, None:
+                    status = task.status
+                    parts = status.message.parts if status.message else []
+                    print_parts(parts, task.status.state.value)
+                    final_status = status.state.value
+                    if status.state.value == "completed":
+                        print(task.artifacts)
+                        artifacts = task.artifacts
+
+                case _:
+                    print("Unhandled event")
+
+        return final_status, artifacts
+
+
+def parse_parts(parts) -> tuple[list, list]:
+    text_parts = []
+    data_parts = []
+
+    for part in parts:
+        if isinstance(part.root, TextPart):
+            try:
+                data_item = json.loads(part.root.text)
+                data_parts.append(data_item)
+            except Exception:
+                text_parts.append(part.root.text.strip())
+        elif isinstance(part.root, DataPart):
+            data_parts.append(part.root.data)
+
+    return text_parts, data_parts
+
+
+async def run_assessment(
+    green_url: str, participants: dict[str, str], assessment_config: dict[str, Any]
+):
+    assessment_request = {"participants": participants, "config": assessment_config}
+    msg = json.dumps(assessment_request)
+
+    status = None
+    artifacts = []
+    try:
+        status, artifacts = await send_message(msg, green_url)
+    except Exception as e:
+        print(f"Assessment failed: {e}")
+        status = "failed"
+    finally:
+        all_data_parts = []
+        for artifact in artifacts or []:
+            _, data_parts = parse_parts(artifact.parts)
+            all_data_parts.extend(data_parts)
+
+        output_data = {"status": status, "results": all_data_parts}
+
+        return output_data

src/config.py

@@ -0,0 +1,35 @@
+import argparse
+import json
+import os
+from dataclasses import dataclass
+
+
+@dataclass
+class Config:
+    port: int
+    green_url: str
+    participants: dict[str, str]  # role -> url
+    assessment_config: dict
+
+
+def load_config() -> Config:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--port", type=int, default=8080)
+    args = parser.parse_args()
+
+    green_url = os.environ.get("GREEN_URL")
+    if not green_url:
+        raise ValueError("GREEN_URL is required")
+
+    participant_urls_raw = os.environ.get("PARTICIPANTS", "{}")
+    participants = json.loads(participant_urls_raw)
+
+    assessment_config_raw = os.environ.get("ASSESSMENT_CONFIG", "{}")
+    assessment_config = json.loads(assessment_config_raw)
+
+    return Config(
+        port=args.port,
+        green_url=green_url,
+        participants=participants,
+        assessment_config=assessment_config,
+    )