feat: A2A (Agent-to-Agent) protocol support

examples/a2a/basic_client_example.py

@@ -0,0 +1,53 @@
+"""
+Example: Calling an A2A agent as a tool from an OpenAI agent.
+
+This example demonstrates using ``A2AClientTool`` to call a remote A2A agent
+as a function tool. It assumes you have an A2A agent running at the given URL.
+
+To try it out:
+
+1. Start a sample A2A server (e.g. the hello-world agent from a2a-sdk):
+   ``cd a2a-python/samples && python cli.py server``
+
+2. Run this script:
+   ``python examples/a2a/basic_client_example.py``
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from agents import Agent, Runner
+from agents.extensions.a2a import A2AClientTool
+
+
+async def main() -> None:
+    # Create a tool that wraps a remote A2A agent.
+    # from_url() fetches the AgentCard at .well-known/agent-card.json
+    research_tool = await A2AClientTool.from_url(
+        url="http://localhost:10000",
+        tool_name="research_agent",
+        tool_description=(
+            "Ask the research agent to find and summarize information. "
+            "Use when you need external knowledge."
+        ),
+    )
+
+    orchestrator = Agent(
+        name="Orchestrator",
+        instructions="You are an orchestrator. Use the research_agent tool for external queries.",
+        tools=[research_tool.as_function_tool()],
+    )
+
+    result = await Runner.run(
+        orchestrator,
+        "What are the latest developments in quantum computing?",
+    )
+    print(f"Final output: {result.final_output}")
+
+    # Clean up
+    await research_tool.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -54,6 +54,7 @@ modal = ["modal==1.3.5"]
 runloop = ["runloop_api_client>=1.16.0,<2.0.0"]
 vercel = ["vercel>=0.5.6,<0.6"]
 s3 = ["boto3>=1.34"]
+a2a = ["a2a-sdk>=1.0.0,<2"]
 temporal = [
     "temporalio==1.26.0",
     "textual>=8.2.3,<8.3",
@@ -164,6 +165,10 @@ ignore_missing_imports = true
 module = ["vercel", "vercel.*"]
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = ["a2a", "a2a.*", "a2a_pb2", "google.protobuf"]
+ignore_missing_imports = true
+
 [tool.coverage.run]
 source = ["src/agents"]
 omit = [

src/agents/extensions/a2a/__init__.py

@@ -0,0 +1,96 @@
+"""
+A2A (Agent-to-Agent) protocol integration for the OpenAI Agents SDK.
+
+This extension enables bidirectional interoperability between OpenAI Agents
+and any A2A-compatible agent (built with any framework, in any language):
+
+- **A2A Client Tool**: Call external A2A agents as tools from your OpenAI agent.
+- **A2A Server Agent**: Expose your OpenAI agent as an A2A service so other
+  agents can call it.
+
+The A2A protocol is defined by Google at https://github.com/google/A2A.
+
+Dependencies
+------------
+This module requires the ``a2a-sdk`` package. Install it with::
+
+    pip install openai-agents[a2a]
+
+or directly::
+
+    pip install a2a-sdk
+"""
+
+from __future__ import annotations
+
+from importlib import import_module
+from typing import TYPE_CHECKING, Any
+
+from agents.extensions.memory._optional_imports import raise_optional_dependency_error
+
+if TYPE_CHECKING:
+    from ._agent_card import generate_agent_card
+    from ._client_tool import A2AClientTool
+    from ._server_executor import A2AServerAgent
+    from ._converter import (
+        a2a_context_to_openai_input,
+        a2a_history_to_openai_input_items,
+        a2a_message_to_openai_input_items,
+        openai_error_to_failed_task,
+        openai_final_output_to_artifacts,
+        openai_items_to_a2a_messages,
+        openai_run_result_to_task,
+        openai_stream_event_to_task_status,
+    )
+
+__all__ = [
+    "A2AClientTool",
+    "A2AServerAgent",
+    "generate_agent_card",
+    "a2a_context_to_openai_input",
+    "a2a_history_to_openai_input_items",
+    "a2a_message_to_openai_input_items",
+    "openai_error_to_failed_task",
+    "openai_final_output_to_artifacts",
+    "openai_items_to_a2a_messages",
+    "openai_run_result_to_task",
+    "openai_stream_event_to_task_status",
+]
+
+_LAZY_EXPORTS: dict[str, tuple[str, tuple[str, str] | None]] = {
+    "A2AClientTool": ("._client_tool", ("a2a-sdk", "a2a")),
+    "A2AServerAgent": ("._server_executor", ("a2a-sdk", "a2a")),
+    "generate_agent_card": ("._agent_card", ("a2a-sdk", "a2a")),
+    "a2a_context_to_openai_input": ("._converter", ("a2a-sdk", "a2a")),
+    "a2a_history_to_openai_input_items": ("._converter", ("a2a-sdk", "a2a")),
+    "a2a_message_to_openai_input_items": ("._converter", ("a2a-sdk", "a2a")),
+    "openai_error_to_failed_task": ("._converter", ("a2a-sdk", "a2a")),
+    "openai_final_output_to_artifacts": ("._converter", ("a2a-sdk", "a2a")),
+    "openai_items_to_a2a_messages": ("._converter", ("a2a-sdk", "a2a")),
+    "openai_run_result_to_task": ("._converter", ("a2a-sdk", "a2a")),
+    "openai_stream_event_to_task_status": ("._converter", ("a2a-sdk", "a2a")),
+}
+
+
+def __getattr__(name: str) -> Any:
+    if name not in _LAZY_EXPORTS:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+
+    module_name, optional_dependency = _LAZY_EXPORTS[name]
+    try:
+        module = import_module(module_name, __name__)
+    except ModuleNotFoundError as e:
+        if optional_dependency is None:
+            raise ImportError(f"Failed to import {name}: {e}") from e
+        dependency_name, extra_name = optional_dependency
+        raise_optional_dependency_error(
+            name,
+            dependency_name=dependency_name,
+            extra_name=extra_name,
+            cause=e,
+        )
+
+    value = getattr(module, name)
+    # Cache for subsequent access.
+    globals()[name] = value
+    return value

src/agents/extensions/a2a/_agent_card.py

@@ -0,0 +1,159 @@
+"""
+AgentCard generator — build an A2A ``AgentCard`` from OpenAI Agent metadata.
+
+This module inspects an OpenAI ``Agent`` instance and produces a compliant
+A2A ``AgentCard`` that describes its name, description, skills (derived from
+tools and handoffs), capabilities, and supported interfaces.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from a2a.types.a2a_pb2 import (  # type: ignore[import-untyped]
+        AgentCapabilities,
+        AgentCard,
+        AgentInterface,
+        AgentProvider,
+        AgentSkill,
+    )
+    from agents.agent import Agent
+
+
+def generate_agent_card(
+    agent: Agent[Any],
+    *,
+    url: str,
+    provider: AgentProvider | None = None,
+    capabilities: AgentCapabilities | None = None,
+    supported_interfaces: list[AgentInterface] | None = None,
+    version: str = "1.0.0",
+) -> AgentCard:
+    """Generate an A2A ``AgentCard`` from an OpenAI ``Agent`` instance.
+
+    The card includes:
+
+    - The agent's ``name`` and ``instructions`` (or ``handoff_description``)
+      as the card's ``description``.
+    - One ``AgentSkill`` per ``FunctionTool`` (and per ``Agent.as_tool()``
+      handoff) with its name, description, and tags.
+    - Default ``input_modes`` / ``output_modes`` set to ``["text"]``.
+
+    Args:
+        agent: The OpenAI ``Agent`` to describe.
+        url: The base URL where the agent will be served.
+        provider: Optional ``AgentProvider`` (organisation metadata).
+        capabilities: Optional ``AgentCapabilities``; defaults to streaming
+            enabled with no push notifications.
+        supported_interfaces: Optional list of ``AgentInterface`` entries;
+            defaults to a single JSON-RPC interface at ``url``.
+        version: Version string for the agent card (default ``"1.0.0"``).
+
+    Returns:
+        A populated A2A ``AgentCard`` protobuf message.
+    """
+    from a2a.types.a2a_pb2 import (  # type: ignore[import-untyped]
+        AgentCapabilities,
+        AgentCard,
+        AgentInterface,
+        AgentSkill,
+    )
+
+    from agents.handoffs import Handoff
+    from agents.tool import FunctionTool
+
+    # -- description --------------------------------------------------------
+    description = agent.handoff_description or ""
+    if not description and agent.instructions:
+        if isinstance(agent.instructions, str):
+            # Truncate very long instructions for the card.
+            description = agent.instructions[:2000]
+        else:
+            description = "Dynamic instructions (callable)."
+
+    # -- skills -------------------------------------------------------------
+    skills: list[AgentSkill] = []
+
+    for tool in agent.tools:
+        if not isinstance(tool, FunctionTool):
+            continue
+        skill = AgentSkill(
+            id=tool.name,
+            name=tool.name,
+            description=tool.description[:2000],
+            tags=_tool_tags(tool),
+            input_modes=["text"],
+            output_modes=["text"],
+        )
+        skills.append(skill)
+
+    for handoff in agent.handoffs:
+        if isinstance(handoff, Handoff):
+            skill = AgentSkill(
+                id=handoff.tool_name,
+                name=handoff.agent_name,
+                description=handoff.handoff_description or handoff.tool_description or "",
+                tags=["handoff"],
+                input_modes=["text"],
+                output_modes=["text"],
+            )
+            skills.append(skill)
+
+    # -- capabilities -------------------------------------------------------
+    if capabilities is None:
+        capabilities = AgentCapabilities(
+            streaming=True,
+            push_notifications=False,
+        )
+
+    # -- interfaces ---------------------------------------------------------
+    if supported_interfaces is None:
+        supported_interfaces = [
+            AgentInterface(
+                url=url,
+                protocol_binding="a2a-json-rpc",
+                protocol_version="1.0",
+            ),
+        ]
+
+    return AgentCard(
+        name=agent.name,
+        description=description,
+        version=version,
+        supported_interfaces=supported_interfaces,
+        default_input_modes=["text"],
+        default_output_modes=["text"],
+        capabilities=capabilities,
+        skills=skills,
+        provider=provider,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _tool_tags(tool: Any) -> list[str]:
+    """Infer A2A skill tags from tool metadata."""
+    tags = ["tool"]
+
+    # Tag known hosted tools
+    tool_type = type(tool).__name__
+    type_to_tag: dict[str, str] = {
+        "FileSearchTool": "file-search",
+        "WebSearchTool": "web-search",
+        "CodeInterpreterTool": "code-interpreter",
+        "HostedMCPTool": "mcp",
+        "ShellTool": "shell",
+    }
+    if tag := type_to_tag.get(tool_type):
+        tags.append(tag)
+
+    # Include namespace if present
+    namespace = getattr(tool, "_tool_namespace", None)
+    if namespace:
+        tags.append(f"namespace:{namespace}")
+
+    return tags