ci: run mandatory and capabilities TCK tests for JSON-RPC transport

.github/workflows/run-tck.yaml

@@ -0,0 +1,112 @@
+name: Run TCK
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+    paths-ignore:
+      - '**.md'
+      - 'LICENSE'
+      - '.github/CODEOWNERS'
+
+env:
+  TCK_VERSION: 0.3.0.beta3
+  SUT_BASE_URL: http://localhost:41241
+  SUT_JSONRPC_URL: http://localhost:41241/a2a/jsonrpc
+  UV_SYSTEM_PYTHON: 1
+  TCK_STREAMING_TIMEOUT: 5.0
+
+concurrency:
+  group: '${{ github.workflow }} @ ${{ github.head_ref || github.ref }}'
+  cancel-in-progress: true
+
+jobs:
+  tck-test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout a2a-python
+        uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: true
+          cache-dependency-glob: "uv.lock"
+
+      - name: Set up Python ${{ matrix.python-version }}
+        run: uv python install ${{ matrix.python-version }}
+
+      - name: Install Dependencies
+        run: uv sync --all-extras --locked
+
+      - name: Checkout a2a-tck
+        uses: actions/checkout@v4
+        with:
+          repository: a2aproject/a2a-tck
+          path: tck/a2a-tck
+          ref: ${{ env.TCK_VERSION }}
+
+      - name: Install TCK dependencies
+        run: |
+          cd tck/a2a-tck
+          pip install uv
+          uv pip install -e .
+
+      - name: Start SUT
+        run: |
+          uv run tck/sut_agent.py &
+        env:
+          HTTP_PORT: 41241
+
+      - name: Wait for SUT to start
+        run: |
+          URL="${{ env.SUT_BASE_URL }}/.well-known/agent-card.json"
+          EXPECTED_STATUS=200
+          TIMEOUT=120
+          RETRY_INTERVAL=2
+          START_TIME=$(date +%s)
+
+          while true; do
+            CURRENT_TIME=$(date +%s)
+            ELAPSED_TIME=$((CURRENT_TIME - START_TIME))
+
+            if [ "$ELAPSED_TIME" -ge "$TIMEOUT" ]; then
+                echo "❌ Timeout: Server did not respond with status $EXPECTED_STATUS within $TIMEOUT seconds."
+                exit 1
+            fi
+
+            HTTP_STATUS=$(curl --output /dev/null --silent --write-out "%{http_code}" "$URL") || true
+            echo "STATUS: ${HTTP_STATUS}"
+
+            if [ "$HTTP_STATUS" -eq "$EXPECTED_STATUS" ]; then
+                echo "✅ Server is up! Received status $HTTP_STATUS after $ELAPSED_TIME seconds."
+                break;
+            fi
+
+            echo "⏳ Server not ready (status: $HTTP_STATUS). Retrying in $RETRY_INTERVAL seconds..."
+            sleep "$RETRY_INTERVAL"
+          done
+
+      - name: Run TCK (mandatory)
+        id: run-tck-mandatory
+        timeout-minutes: 5
+        run: |
+          ./run_tck.py --sut-url ${{ env.SUT_JSONRPC_URL }} --category mandatory --transports jsonrpc
+        working-directory: tck/a2a-tck
+
+      - name: Run TCK (capabilities)
+        id: run-tck-capabilities
+        timeout-minutes: 5
+        run: |
+          ./run_tck.py --sut-url ${{ env.SUT_JSONRPC_URL }} --category capabilities --transports jsonrpc
+        working-directory: tck/a2a-tck
+
+      - name: Stop SUT
+        if: always()
+        run: |
+           pkill -f sut_agent.py || true
+          sleep 2

tck/sut_agent.py

@@ -0,0 +1,189 @@
+
+import asyncio
+import logging
+import os
+import uuid
+from datetime import datetime, timezone
+
+from datetime import datetime, timezone
+
+from fastapi import FastAPI
+from uvicorn import Config, Server
+
+
+from a2a.server.agent_execution.agent_executor import AgentExecutor
+from a2a.server.agent_execution.context import RequestContext
+from a2a.server.events.event_queue import EventQueue
+from a2a.server.events.in_memory_queue_manager import InMemoryQueueManager
+from a2a.server.request_handlers.default_request_handler import DefaultRequestHandler
+from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
+from a2a.server.apps.jsonrpc.fastapi_app import A2AFastAPIApplication
+from a2a.types import (
+    AgentCard,
+    AgentCapabilities,
+    AgentProvider,
+    Message,
+    TextPart,
+    Task,
+    TaskState,
+    TaskStatus,
+    TaskStatusUpdateEvent,
+)
+from a2a.auth.user import UnauthenticatedUser
+from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("SUTAgent")
+
+class SUTAgentExecutor(AgentExecutor):
+    def __init__(self):
+        self.running_tasks = set()
+        self.last_context_id = None
+
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        api_task_id = context.task_id
+        if api_task_id in self.running_tasks:
+            self.running_tasks.remove(api_task_id)
+
+        status_update = TaskStatusUpdateEvent(
+            task_id=api_task_id,
+            context_id=self.last_context_id or str(uuid.uuid4()),
+            status=TaskStatus(
+                state=TaskState.canceled,
+                timestamp=datetime.now(timezone.utc).isoformat(),
+            ),
+            final=True,
+        )
+        await event_queue.enqueue_event(status_update)
+
+    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
+        user_message = context.message
+        task_id = context.task_id
+        context_id = context.context_id
+        self.last_context_id = context_id
+
+        self.running_tasks.add(task_id)
+
+        logger.info(
+            f"[SUTAgentExecutor] Processing message {user_message.message_id} "
+            f"for task {task_id} (context: {context_id})"
+        )
+
+        working_status = TaskStatusUpdateEvent(
+            task_id=task_id,
+            context_id=context_id,
+            status=TaskStatus(
+                state=TaskState.working,
+                message=Message(
+                    role="agent",
+                    message_id=str(uuid.uuid4()),
+                    parts=[TextPart(text="Processing your question")],
+                    task_id=task_id,
+                    context_id=context_id,
+                ),
+                timestamp=datetime.now(timezone.utc).isoformat(),
+            ),
+            final=False,
+        )
+        await event_queue.enqueue_event(working_status)
+
+        agent_reply_text = "Hello world!"
+        await asyncio.sleep(3)  # Simulate processing delay
+
+        if task_id not in self.running_tasks:
+            logger.info(f"Task {task_id} was cancelled.")
+            return
+
+        logger.info(f"[SUTAgentExecutor] Response: {agent_reply_text}")
+
+        agent_message = Message(
+            role="agent",
+            message_id=str(uuid.uuid4()),
+            parts=[TextPart(text=agent_reply_text)],
+            task_id=task_id,
+            context_id=context_id,
+        )
+
+        final_update = TaskStatusUpdateEvent(
+            task_id=task_id,
+            context_id=context_id,
+            status=TaskStatus(
+                state=TaskState.input_required,
+                message=agent_message,
+                timestamp=datetime.now(timezone.utc).isoformat(),
+            ),
+            final=True,
+        )
+        await event_queue.enqueue_event(final_update)
+
+
+
+async def main():
+    HTTP_PORT = int(os.environ.get("HTTP_PORT", 41241))
+
+    # 1. Setup Executor and Handlers
+    agent_executor = SUTAgentExecutor()
+    task_store = InMemoryTaskStore()
+    queue_manager = InMemoryQueueManager()
+
+    request_handler = DefaultRequestHandler(
+        task_store=task_store,
+        queue_manager=queue_manager,
+        agent_executor=agent_executor,
+    )
+
+    # 2. Create Agent Card (JSON-RPC only)
+    sut_agent_card = AgentCard(
+        name="SUT Agent",
+        description="A sample agent to be used as SUT against tck tests.",
+        url=f"http://localhost:{HTTP_PORT}/a2a/jsonrpc",
+        provider=AgentProvider(
+            organization="A2A Samples",
+            url="https://example.com/a2a-samples",
+        ),
+        version="1.0.0",
+        protocol_version="0.3.0",
+        capabilities=AgentCapabilities(
+            streaming=True,
+            push_notifications=False,
+            state_transition_history=True,
+        ),
+        default_input_modes=["text"],
+        default_output_modes=["text", "task-status"],
+        skills=[
+            {
+                "id": "sut_agent",
+                "name": "SUT Agent",
+                "description": "Simulate the general flow of a streaming agent.",
+                "tags": ["sut"],
+                "examples": ["hi", "hello world", "how are you", "goodbye"],
+                "input_modes": ["text"],
+                "output_modes": ["text", "task-status"],
+            }
+        ],
+        supports_authenticated_extended_card=False,
+        preferred_transport="JSONRPC",
+        additional_interfaces=[
+            {"url": f"http://localhost:{HTTP_PORT}/a2a/jsonrpc", "transport": "JSONRPC"},
+        ],
+    )
+
+    # 3. Setup HTTP App
+    json_rpc_app = A2AFastAPIApplication(
+        agent_card=sut_agent_card,
+        http_handler=request_handler,
+    )
+    app = json_rpc_app.build(
+        rpc_url="/a2a/jsonrpc",
+        agent_card_url="/.well-known/agent-card.json"
+    )
+
+    logger.info(f"Starting HTTP server on port {HTTP_PORT}...")
+    config = Config(app, host="0.0.0.0", port=HTTP_PORT, log_level="info")
+    server = Server(config)
+
+    await server.serve()
+
+if __name__ == "__main__":
+    asyncio.run(main())