Add template (#1)

* Add template

* Update naming in Dockerfile

* Remove unused imports; improve comments

Author: staverm

.github/workflows/publish.yml

@@ -0,0 +1,74 @@
+name: Publish Agent
+
+# Trigger this workflow when pushing main branch and tags
+on:
+  push:
+    branches:
+      - main
+    tags:
+      - 'v*'  # Trigger on version tags like v1.0.0, v1.1.0
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+
+    # These permissions are required for the workflow to:
+    # - Read repository contents (checkout code)
+    # - Write to GitHub Container Registry (push Docker images)
+    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 }}
+          # GITHUB_TOKEN is automatically provided by GitHub Actions
+          # No manual secret configuration needed!
+          # It has permissions based on the 'permissions' block above
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata for Docker
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: ghcr.io/${{ github.repository }}
+          tags: |
+            # For tags like v1.0, create tag '1.0'
+            type=semver,pattern={{version}}
+            # For tags like v1.0, create tag '1'
+            type=semver,pattern={{major}}
+            # For main branch, create tag 'latest'
+            type=raw,value=latest,enable={{is_default_branch}}
+            # For PRs, create tag 'pr-123'
+            type=ref,event=pr
+
+      - name: Build and push Docker image
+        id: build
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          file: Dockerfile
+          # Only push if this is a push event (not a PR)
+          # PRs will build but not push to avoid polluting the registry
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          # Explicitly build for linux/amd64 (GitHub Actions default)
+          platforms: linux/amd64
+
+      - name: Output image digest
+        if: github.event_name != 'pull_request'
+        run: |
+          echo "## Docker Image Published :rocket:" >> $GITHUB_STEP_SUMMARY
+          echo "" >> $GITHUB_STEP_SUMMARY
+          echo "**Tags:** ${{ steps.meta.outputs.tags }}" >> $GITHUB_STEP_SUMMARY
+          echo "" >> $GITHUB_STEP_SUMMARY
+          echo "**Digest:** \`${{ steps.build.outputs.digest }}\`" >> $GITHUB_STEP_SUMMARY
+          echo "" >> $GITHUB_STEP_SUMMARY
+          echo "Use this digest in your MANIFEST.json for reproducibility." >> $GITHUB_STEP_SUMMARY

.gitignore

@@ -0,0 +1,14 @@
+.DS_Store
+.env
+.python-version
+
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv

Dockerfile

@@ -0,0 +1,16 @@
+FROM ghcr.io/astral-sh/uv:python3.13-bookworm
+
+RUN adduser agent
+USER agent
+WORKDIR /home/agent
+
+COPY pyproject.toml uv.lock README.md ./
+COPY src src
+
+RUN \
+    --mount=type=cache,target=/home/agent/.cache/uv,uid=1000 \
+    uv sync --locked
+
+ENTRYPOINT ["uv", "run", "src/server.py"]
+CMD ["--host", "0.0.0.0"]
+EXPOSE 9009

README.md

@@ -0,0 +1,59 @@
+# A2A Agent Template
+
+A minimal template for building [A2A (Agent-to-Agent)](https://a2a-protocol.org/latest/) agents.
+
+## Project Structure
+
+```
+src/
+├─ server.py      # Server setup and agent card configuration
+├─ executor.py    # A2A request handling
+├─ agent.py       # Your agent implementation goes here
+└─ messenger.py   # A2A messaging utilities
+Dockerfile        # Docker configuration
+pyproject.toml    # Python dependencies
+uv.lock           # Locked dependencies
+```
+
+## Getting Started
+
+1. **Create your repository** - Click "Use this template" to create your own repository from this template
+
+2. **Implement your agent** - Add your agent logic to the `run` method in [`src/agent.py`](src/agent.py)
+
+3. **Configure your agent card** - Fill in your agent's metadata (name, skills, description) in [`src/server.py`](src/server.py)
+
+## Running Locally
+
+```bash
+# Install dependencies
+uv sync
+
+# Run the server
+uv run src/server.py
+```
+
+## Running with Docker
+
+```bash
+# Build the image
+docker build -t my-agent .
+
+# Run the container
+docker run -p 9009:9009 my-agent
+```
+
+## Publishing
+
+The repository includes a GitHub Actions workflow that automatically builds and publishes a Docker image of your agent to GitHub Container Registry:
+
+- **Push to `main`** → publishes `latest` tag:
+```
+ghcr.io/<your-username>/<your-repo-name>:latest
+```
+
+- **Create a git tag** (e.g. `git tag v1.0.0 && git push origin v1.0.0`) → publishes version tags:
+```
+ghcr.io/<your-username>/<your-repo-name>:1.0.0
+ghcr.io/<your-username>/<your-repo-name>:1
+```

pyproject.toml

@@ -0,0 +1,10 @@
+[project]
+name = "agent-template"
+version = "0.1.0"
+description = "A template for A2A agents"
+readme = "README.md"
+requires-python = ">=3.13"
+dependencies = [
+    "a2a-sdk[http-server]>=0.3.20",
+    "uvicorn>=0.38.0",
+]

src/agent.py

@@ -0,0 +1,19 @@
+from a2a.server.tasks import TaskUpdater
+from messenger import Messenger
+
+
+class Agent:
+    def __init__(self):
+        self.messenger = Messenger()
+        # initialize other state here
+
+    async def run(self, input_text: str, updater: TaskUpdater) -> None:
+        """Implement your agent logic here.
+
+        Args:
+            input_text: The incoming message text
+            updater: Report progress (update_status) and results (add_artifact)
+
+        Use self.messenger.talk_to_agent(message, url) to call other agents.
+        """
+        raise NotImplementedError("Agent not implemented.")

src/executor.py

@@ -0,0 +1,58 @@
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.server.tasks import TaskUpdater
+from a2a.types import (
+    Task,
+    UnsupportedOperationError,
+    InvalidRequestError,
+)
+from a2a.utils.errors import ServerError
+from a2a.utils import (
+    new_agent_text_message,
+    new_task,
+)
+
+from agent import Agent
+
+
+TERMINAL_STATES = ["completed", "canceled", "rejected", "failed"]
+
+
+class Executor(AgentExecutor):
+    def __init__(self):
+        self.agent_store: dict[str, Agent] = {}
+
+    async def execute(
+        self,
+        context: RequestContext,
+        event_queue: EventQueue,
+    ) -> None:
+        msg = context.message
+        if not msg:
+            raise ServerError(error=InvalidRequestError(message="Missing message in request"))
+
+        task = context.current_task
+        if not task or task.status.state in TERMINAL_STATES:
+            task = new_task(msg)
+
+        context_id = task.context_id
+        agent = self.agent_store.get(context_id)
+        if not agent:
+            agent = Agent()
+            self.agent_store[context_id] = agent
+
+        await event_queue.enqueue_event(task)
+        updater = TaskUpdater(event_queue, task.id, context_id)
+        await updater.start_work(new_agent_text_message(text="Thinking...", context_id=context_id, task_id=task.id))
+
+        try:
+            await agent.run(context.get_user_input(), updater)
+            await updater.complete()
+        except Exception as e:
+            print(f"Task failed with agent error: {e}")
+            await updater.failed(new_agent_text_message(f"Agent error: {e}", context_id=context_id))
+
+    async def cancel(
+        self, request: RequestContext, event_queue: EventQueue
+    ) -> Task | None:
+        raise ServerError(error=UnsupportedOperationError())

src/messenger.py

@@ -0,0 +1,110 @@
+import json
+from uuid import uuid4
+
+import httpx
+from a2a.client import (
+    A2ACardResolver,
+    ClientConfig,
+    ClientFactory,
+    Consumer,
+)
+from a2a.types import (
+    Message,
+    Part,
+    Role,
+    TextPart,
+    DataPart,
+)
+
+
+DEFAULT_TIMEOUT = 300
+
+
+def create_message(*, role: Role = Role.user, text: str, context_id: str | None = None) -> Message:
+    return Message(
+        kind="message",
+        role=role,
+        parts=[Part(TextPart(kind="text", text=text))],
+        message_id=uuid4().hex,
+        context_id=context_id
+    )
+
+def merge_parts(parts: list[Part]) -> str:
+    chunks = []
+    for part in parts:
+        if isinstance(part.root, TextPart):
+            chunks.append(part.root.text)
+        elif isinstance(part.root, DataPart):
+            chunks.append(json.dumps(part.root.data, indent=2))
+    return "\n".join(chunks)
+
+async def send_message(message: str, base_url: str, context_id: str | None = None, streaming=False, consumer: Consumer | None = None):
+    """Returns dict with context_id, response and status (if exists)"""
+    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=streaming,
+        )
+        factory = ClientFactory(config)
+        client = factory.create(agent_card)
+        if consumer:
+            await client.add_event_consumer(consumer)
+
+        outbound_msg = create_message(text=message, context_id=context_id)
+        last_event = None
+        outputs = {
+            "response": "",
+            "context_id": None
+        }
+
+        # if streaming == False, only one event is generated
+        async for event in client.send_message(outbound_msg):
+            last_event = event
+
+        match last_event:
+            case Message() as msg:
+                outputs["context_id"] = msg.context_id
+                outputs["response"] += merge_parts(msg.parts)
+
+            case (task, update):
+                outputs["context_id"] = task.context_id
+                outputs["status"] = task.status.state.value
+                msg = task.status.message
+                if msg:
+                    outputs["response"] += merge_parts(msg.parts)
+                if task.artifacts:
+                    for artifact in task.artifacts:
+                        outputs["response"] += merge_parts(artifact.parts)
+
+            case _:
+                pass
+
+        return outputs
+
+
+class Messenger:
+    def __init__(self):
+        self._context_ids = {}
+
+    async def talk_to_agent(self, message: str, url: str, new_conversation: bool = False):
+        """
+        Communicate with another agent by sending a message and receiving their response.
+
+        Args:
+            message: The message to send to the agent
+            url: The agent's URL endpoint
+            new_conversation: If True, start fresh conversation; if False, continue existing conversation
+
+        Returns:
+            str: The agent's response message
+        """
+        outputs = await send_message(message=message, base_url=url, context_id=None if new_conversation else self._context_ids.get(url, None))
+        if outputs.get("status", "completed") != "completed":
+            raise RuntimeError(f"{url} responded with: {outputs}")
+        self._context_ids[url] = outputs.get("context_id", None)
+        return outputs["response"]
+
+    def reset(self):
+        self._context_ids = {}