feat: add AG2 (formerly AutoGen) integration package and sample

.claude/skills/kagent/SKILL.md

@@ -61,7 +61,7 @@ For Helm install, other LLM providers, and provider-specific configuration, see
 
 kagent uses Kubernetes CRDs to manage agents, models, and tools:
 
-- **Agent** (`kagent.dev/v1alpha2`) — Defines an AI agent. Two types: **Declarative** (YAML-defined, controller-managed) and **BYO** (custom container image with any framework: Google ADK, OpenAI Agents SDK, LangGraph, CrewAI).
+- **Agent** (`kagent.dev/v1alpha2`) — Defines an AI agent. Two types: **Declarative** (YAML-defined, controller-managed) and **BYO** (custom container image with any framework: Google ADK, AG2, OpenAI Agents SDK, LangGraph, CrewAI).
 - **ModelConfig** (`kagent.dev/v1alpha2`) — Configures LLM provider and model. Agents reference a ModelConfig by name.
 - **RemoteMCPServer** (`kagent.dev/v1alpha2`) — Connects agents to external MCP tool servers via HTTP.
 - **MCPServer** (KMCP) — Deploys and manages MCP server pods in the cluster.

CLAUDE.md

@@ -68,7 +68,7 @@ kagent/
 | Language | Use For | Don't Use For |
 |----------|---------|---------------|
 | **Go** | K8s controllers, CLI tools, core APIs, HTTP server, database layer | Agent runtime, LLM integrations, UI |
-| **Python** | Agent runtime, ADK, LLM integrations, AI/ML logic | Kubernetes controllers, CLI, infrastructure |
+| **Python** | Agent runtime, ADK, AG2, LLM integrations, AI/ML logic | Kubernetes controllers, CLI, infrastructure |
 | **TypeScript** | Web UI components and API clients only | Backend logic, controllers, agents |
 
 **Rule of thumb:** Infrastructure in Go, AI/Agent logic in Python, User interface in TypeScript.

README.md

@@ -94,7 +94,7 @@ Kagent has 4 core components:
 
 - **Controller**: The controller is a Kubernetes controller that watches the kagent custom resources and creates the necessary resources to run the agents.
 - **UI**: The UI is a web UI that allows you to manage the agents and tools.
-- **Engine**: The engine runs your agents using [ADK](https://google.github.io/adk-docs/).
+- **Engine**: The engine runs your agents using [ADK](https://google.github.io/adk-docs/), [AG2](https://docs.ag2.ai) (formerly AutoGen), and other supported frameworks.
 - **CLI**: The CLI is a command-line tool that allows you to manage the agents and tools.
 
 ## Get Involved

contrib/cncf/technical-review.md

@@ -40,7 +40,7 @@ The primary use case is enabling AI-powered automation and intelligent operation
 
 - Multi-agent coordination for complex operational workflows (via A2A protocol)
 - Integration with service mesh (Istio), observability (Prometheus/Grafana), and deployment tools (Helm, Argo Rollouts)
-- Custom agent development using multiple frameworks (ADK, CrewAI, LangGraph)
+- Custom agent development using multiple frameworks (ADK, AG2, CrewAI, LangGraph)
 
 **Unsupported Use Cases:**
 

examples/modelconfig-with-tls.yaml

@@ -297,7 +297,7 @@ metadata:
   name: internal-assistant
   namespace: kagent
 spec:
-  # Framework selection (ADK, LangGraph, or CrewAI)
+  # Framework selection (ADK, AG2, LangGraph, or CrewAI)
   framework: ADK
 
   # Reference to ModelConfig with TLS configuration

python/packages/kagent-ag2/pyproject.toml

@@ -0,0 +1,18 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "kagent-ag2"
+version = "0.1.0"
+requires-python = ">=3.11"
+dependencies = [
+    "kagent-core>=0.1.0",
+    "ag2[openai]>=0.11.0",
+    "a2a-sdk>=0.3.23",
+    "fastapi>=0.115.0",
+    "uvicorn>=0.34.0",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/kagent"]

python/packages/kagent-ag2/src/kagent/ag2/__init__.py

@@ -0,0 +1,7 @@
+"""AG2 (formerly AutoGen) integration for kagent."""
+
+from ._a2a import KAgentApp
+
+__version__ = "0.1.0"
+
+__all__ = ["KAgentApp"]

python/packages/kagent-ag2/src/kagent/ag2/_a2a.py

@@ -0,0 +1,133 @@
+"""KAgentApp for AG2 — builds a FastAPI application."""
+
+import logging
+from collections.abc import Callable
+
+from a2a.server.request_handling import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore
+from a2a.types import AgentCard
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from kagent.core import (
+    KAgentConfig,
+    KAgentRequestContextBuilder,
+    KAgentTaskStore,
+    configure_tracing,
+)
+
+from ._executor import AG2AgentExecutor
+
+try:
+    from a2a.server.apps import A2AFastAPIApplication
+except ImportError:
+    from a2a.server.apps import A2AStarletteApplication as A2AFastAPIApplication
+
+logger = logging.getLogger(__name__)
+
+
+class KAgentApp:
+    """Builds a FastAPI app wrapping an AG2 multi-agent group chat.
+
+    Args:
+        pattern_factory: Callable returning a fresh AG2 Pattern
+            per request.
+        agent_card: A2A AgentCard dict or AgentCard instance.
+        max_rounds: Max conversation rounds per request.
+        config: Optional KAgentConfig (reads from env if None).
+    """
+
+    def __init__(
+        self,
+        pattern_factory: Callable,
+        agent_card: dict | AgentCard,
+        max_rounds: int = 20,
+        config: KAgentConfig | None = None,
+    ):
+        self._pattern_factory = pattern_factory
+        self._max_rounds = max_rounds
+        self._config = config or KAgentConfig()
+
+        if isinstance(agent_card, dict):
+            self._agent_card = AgentCard(**agent_card)
+        else:
+            self._agent_card = agent_card
+
+    def build(self) -> FastAPI:
+        """Build and return the FastAPI application."""
+        executor = AG2AgentExecutor(
+            pattern_factory=self._pattern_factory,
+            max_rounds=self._max_rounds,
+        )
+
+        # Use persistent task store if kagent backend is
+        # available, otherwise in-memory
+        try:
+            task_store = KAgentTaskStore(
+                url=self._config.url,
+                app_name=self._config.app_name,
+            )
+        except (ConnectionError, OSError, ValueError) as e:
+            logger.warning(
+                "kagent backend not available (%s), using "
+                "in-memory task store",
+                e,
+            )
+            task_store = InMemoryTaskStore()
+
+        request_handler = DefaultRequestHandler(
+            agent_executor=executor,
+            task_store=task_store,
+            context_builder=KAgentRequestContextBuilder(),
+        )
+
+        a2a_app = A2AFastAPIApplication(
+            agent_card=self._agent_card,
+            http_handler=request_handler,
+        )
+
+        app = FastAPI(title=self._agent_card.name)
+
+        configure_tracing(self._config.name, self._config.namespace, app)
+
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=["*"],
+            allow_methods=["*"],
+            allow_headers=["*"],
+        )
+
+        @app.get("/health")
+        async def health():
+            return {"status": "ok"}
+
+        a2a_app.mount(app)
+        return app
+
+    def build_local(self) -> FastAPI:
+        """Build app with in-memory task store (no kagent
+        backend required)."""
+        executor = AG2AgentExecutor(
+            pattern_factory=self._pattern_factory,
+            max_rounds=self._max_rounds,
+        )
+
+        task_store = InMemoryTaskStore()
+        request_handler = DefaultRequestHandler(
+            agent_executor=executor,
+            task_store=task_store,
+        )
+
+        a2a_app = A2AFastAPIApplication(
+            agent_card=self._agent_card,
+            http_handler=request_handler,
+        )
+
+        app = FastAPI(title=self._agent_card.name)
+
+        @app.get("/health")
+        async def health():
+            return {"status": "ok"}
+
+        a2a_app.mount(app)
+        return app

python/packages/kagent-ag2/src/kagent/ag2/_executor.py

@@ -0,0 +1,121 @@
+"""AG2 agent executor for the A2A protocol."""
+
+import asyncio
+import logging
+from collections.abc import Callable
+
+try:
+    from typing import override  # Python 3.12+
+except ImportError:
+    from typing_extensions import override
+
+from a2a.server.agent_execution import AgentExecutor
+from a2a.server.events import EventQueue
+from a2a.server.request_handling import RequestContext
+from a2a.types import (
+    Part,
+    TextPart,
+)
+from a2a.utils import new_agent_text_message
+
+from autogen.agentchat import initiate_group_chat
+from autogen.agentchat.group.patterns.pattern import Pattern
+
+logger = logging.getLogger(__name__)
+
+
+def _extract_text(context: RequestContext) -> str:
+    """Extract text content from an A2A request."""
+    if context.message and context.message.parts:
+        for part in context.message.parts:
+            if isinstance(part, Part) and isinstance(
+                part.root, TextPart
+            ):
+                return part.root.text
+            if isinstance(part, TextPart):
+                return part.text
+    return ""
+
+
+class AG2AgentExecutor(AgentExecutor):
+    """Wraps an AG2 multi-agent group chat as an A2A executor.
+
+    Args:
+        pattern_factory: Callable that returns a fresh Pattern
+            for each request (avoids state leakage between
+            requests).
+        max_rounds: Maximum conversation rounds per request.
+    """
+
+    def __init__(
+        self,
+        pattern_factory: Callable[[], Pattern],
+        max_rounds: int = 20,
+    ):
+        super().__init__()
+        self._pattern_factory = pattern_factory
+        self._max_rounds = max_rounds
+
+    @override
+    async def execute(
+        self,
+        context: RequestContext,
+        event_queue: EventQueue,
+    ) -> None:
+        message = _extract_text(context)
+        if not message:
+            await event_queue.enqueue_event(
+                new_agent_text_message(
+                    "Error: No message content received."
+                )
+            )
+            return
+
+        # Signal that work has started
+        await event_queue.enqueue_event(
+            new_agent_text_message("Processing request...")
+        )
+
+        try:
+            # Run AG2 group chat in a thread (sync -> async)
+            pattern = self._pattern_factory()
+            result, ctx, last_agent = await asyncio.to_thread(
+                initiate_group_chat,
+                pattern=pattern,
+                messages=message,
+                max_rounds=self._max_rounds,
+            )
+
+            # Extract final response
+            final_message = ""
+            for msg in reversed(result.chat_history):
+                content = msg.get("content", "")
+                if content and "TERMINATE" not in content:
+                    final_message = content
+                    break
+
+            if not final_message:
+                final_message = "Research complete."
+
+            await event_queue.enqueue_event(
+                new_agent_text_message(final_message)
+            )
+
+        except Exception as e:
+            logger.exception("AG2 group chat failed")
+            await event_queue.enqueue_event(
+                new_agent_text_message(f"Error: {e}")
+            )
+
+    @override
+    async def cancel(
+        self,
+        context: RequestContext,
+        event_queue: EventQueue,
+    ) -> None:
+        await event_queue.enqueue_event(
+            new_agent_text_message(
+                "Cancellation is not supported for AG2 "
+                "group chat conversations."
+            )
+        )

python/packages/kagent-ag2/tests/test_executor.py

@@ -0,0 +1,42 @@
+"""Basic tests for AG2AgentExecutor."""
+
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+
+from kagent.ag2._executor import AG2AgentExecutor, _extract_text
+
+
+def test_extract_text_empty_parts():
+    ctx = MagicMock()
+    ctx.message.parts = []
+    assert _extract_text(ctx) == ""
+
+
+def test_extract_text_no_message():
+    ctx = MagicMock()
+    ctx.message = None
+    result = _extract_text(ctx)
+    assert result == ""
+
+
+def test_executor_instantiation():
+    factory = MagicMock()
+    executor = AG2AgentExecutor(pattern_factory=factory, max_rounds=5)
+    assert executor._max_rounds == 5
+    assert executor._pattern_factory is factory
+
+
+@pytest.mark.asyncio
+async def test_execute_empty_message():
+    factory = MagicMock()
+    executor = AG2AgentExecutor(pattern_factory=factory, max_rounds=5)
+
+    ctx = MagicMock()
+    ctx.message.parts = []
+    queue = AsyncMock()
+
+    await executor.execute(ctx, queue)
+
+    # Should have sent an error message, not called the factory
+    factory.assert_not_called()
+    queue.enqueue_event.assert_called()

python/samples/ag2/research-report/Dockerfile

@@ -0,0 +1,5 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY . .
+RUN pip install --no-cache-dir .
+CMD ["python", "main.py"]

python/samples/ag2/research-report/agent-card.json

@@ -0,0 +1,17 @@
+{
+    "name": "AG2 Research Report",
+    "description": "A multi-agent research team powered by AG2 (formerly AutoGen). Two specialists (researcher + analyst) collaborate to produce structured reports on any topic.",
+    "url": "http://localhost:8080",
+    "version": "0.1.0",
+    "capabilities": {
+        "streaming": false,
+        "pushNotifications": false
+    },
+    "skills": [
+        {
+            "id": "research-report",
+            "name": "Research Report",
+            "description": "Produce a structured research report on a given topic"
+        }
+    ]
+}