feat(a2a): add secure peer protocol integration

MANIFEST.in

@@ -7,6 +7,8 @@ graft locales
 # built from the sdist (e.g. Homebrew, downstream packagers). package-data
 # below covers the wheel; this covers the sdist. See #34034 / #28149.
 recursive-include plugins plugin.yaml plugin.yml
+# Plugin-local opt-in skills (for example a2a-platform:a2a-peer).
+recursive-include plugins SKILL.md
 # Gateway assets include images plus YAML catalogs such as status_phrases.yaml.
 recursive-include gateway/assets *
 global-exclude __pycache__

agent/agent_init.py

@@ -346,6 +346,7 @@ def init_agent(
     checkpoint_max_total_size_mb: int = 500,
     checkpoint_max_file_size_mb: int = 10,
     pass_session_id: bool = False,
+    agent_tool_policy: str = "configured",
 ):
     """
     Initialize the AI Agent.
@@ -624,6 +625,11 @@ def init_agent(
     # Store toolset filtering options
     agent.enabled_toolsets = enabled_toolsets
     agent.disabled_toolsets = disabled_toolsets
+    agent.agent_tool_policy = str(agent_tool_policy or "configured").strip().lower()
+    if agent.agent_tool_policy not in {"configured", "explicit", "none"}:
+        raise ValueError(
+            "agent_tool_policy must be one of: configured, explicit, none"
+        )
 
     # Model response configuration
     agent.max_tokens = max_tokens  # None = use model default
@@ -677,7 +683,7 @@ def init_agent(
     # Opt-out flag for the between-turns MCP tool refresh (build_turn_context).
     # Set on internal forks (e.g. background_review) that must keep ``tools[]``
     # byte-identical to a parent for provider cache parity.
-    agent._skip_mcp_refresh = False
+    agent._skip_mcp_refresh = agent.agent_tool_policy == "none"
     # Registry generation the current tool snapshot was derived from. Lets a
     # late/concurrent refresh reject a stale (older-generation) rebuild instead
     # of clobbering a newer one. Set adjacent to the tool snapshot below.
@@ -1196,6 +1202,7 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
         enabled_toolsets=enabled_toolsets,
         disabled_toolsets=disabled_toolsets,
         quiet_mode=agent.quiet_mode,
+        agent_tool_policy=agent.agent_tool_policy,
     )
 
     # Show tool configuration and store valid tool names for validation
@@ -1482,7 +1489,8 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
             agent._memory_manager = None
 
     from agent.memory_manager import inject_memory_provider_tools as _inject_memory_provider_tools
-    _inject_memory_provider_tools(agent)
+    if agent.agent_tool_policy != "none":
+        _inject_memory_provider_tools(agent)
 
     # Skills config: nudge interval for skill creation reminders
     agent._skill_nudge_interval = 10
@@ -1958,7 +1966,8 @@ def _moa_reference_relay(event: str, **kwargs: Any) -> None:
     # same local-model latency penalty.
     agent._context_engine_tool_names: set = set()
     if (
-        hasattr(agent, "context_compressor")
+        agent.agent_tool_policy != "none"
+        and hasattr(agent, "context_compressor")
         and agent.context_compressor
         and agent.tools is not None
         and (

agent/agent_runtime_helpers.py

@@ -2413,6 +2413,7 @@ def _execute(next_args: dict) -> Any:
                 skip_tool_request_middleware=True,
                 enabled_toolsets=getattr(agent, "enabled_toolsets", None),
                 disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                agent_tool_policy=getattr(agent, "agent_tool_policy", "configured"),
                 tool_request_middleware_trace=list(_tool_middleware_trace),
             )
 

agent/background_review.py

@@ -723,6 +723,7 @@ def _bg_review_auto_deny(command, description, **kwargs):
                 parent_session_id=agent.session_id,
                 enabled_toolsets=getattr(agent, "enabled_toolsets", None),
                 disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                agent_tool_policy=getattr(agent, "agent_tool_policy", "configured"),
                 skip_memory=True,
                 **_fork_kwargs,
             )

agent/tool_executor.py

@@ -240,10 +240,14 @@ def _tool_search_scoped_names(agent) -> frozenset:
 
     enabled = getattr(agent, "enabled_toolsets", None)
     disabled = getattr(agent, "disabled_toolsets", None)
+    policy = str(
+        getattr(agent, "agent_tool_policy", "configured") or "configured"
+    ).strip().lower()
     cache_key = (
         getattr(_registry, "_generation", 0),
         frozenset(enabled) if enabled is not None else None,
         frozenset(disabled) if disabled is not None else None,
+        policy,
     )
     cached = getattr(agent, "_tool_search_scope_cache", None)
     if cached is not None and cached[0] == cache_key:
@@ -254,6 +258,7 @@ def _tool_search_scoped_names(agent) -> frozenset:
             disabled_toolsets=disabled,
             quiet_mode=True,
             skip_tool_search_assembly=True,
+            agent_tool_policy=policy,
         ) or []
         names = _ts.scoped_deferrable_names(scoped_defs)
     except Exception:
@@ -1494,6 +1499,7 @@ def _execute(next_args: dict) -> Any:
                     skip_tool_request_middleware=True,
                     enabled_toolsets=getattr(agent, "enabled_toolsets", None),
                     disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                    agent_tool_policy=getattr(agent, "agent_tool_policy", "configured"),
                     tool_request_middleware_trace=list(middleware_trace),
                 )
                 _spinner_result = function_result
@@ -1536,6 +1542,7 @@ def _execute(next_args: dict) -> Any:
                     skip_tool_request_middleware=True,
                     enabled_toolsets=getattr(agent, "enabled_toolsets", None),
                     disabled_toolsets=getattr(agent, "disabled_toolsets", None),
+                    agent_tool_policy=getattr(agent, "agent_tool_policy", "configured"),
                     tool_request_middleware_trace=list(middleware_trace),
                 )
             except KeyboardInterrupt:

gateway/config.py

@@ -322,6 +322,7 @@ def _scan_bundled_plugin_platforms(cls) -> set:
 # dashboard's pre-write mutation validation (hermes_cli/web_server.py) so
 # the two policies cannot drift. Stored as platform .value strings.
 PORT_BINDING_PLATFORM_VALUES = frozenset({
+    "a2a",
     "webhook",
     "api_server",
     "msgraph_webhook",

gateway/platform_registry.py

@@ -30,11 +30,20 @@
 
 import logging
 from dataclasses import dataclass, field
+from enum import Enum
 from typing import Any, Awaitable, Callable, Optional
 
 logger = logging.getLogger(__name__)
 
 
+class AgentToolPolicy(str, Enum):
+    """Host-enforced agent tool policy for a platform adapter."""
+
+    CONFIGURED = "configured"
+    EXPLICIT = "explicit"
+    NONE = "none"
+
+
 @dataclass
 class PlatformEntry:
     """Metadata and factory for a single platform adapter."""
@@ -110,6 +119,16 @@ class PlatformEntry:
     # Do not use markdown.").  Empty string = no hint.
     platform_hint: str = ""
 
+    # Host-enforced tool policy for agents serving this platform. ``explicit``
+    # uses only the platform's saved allowlist; ``none`` is an absolute deny.
+    # Neither policy permits worker context, plugins, MCP refresh, or post-build
+    # injectors to widen the selected surface.
+    agent_tool_policy: AgentToolPolicy | str = AgentToolPolicy.CONFIGURED
+
+    # Whether inbound @file/@folder/@diff-style references may read and expand
+    # local context before the message reaches the agent.
+    inbound_context_references_enabled: bool = True
+
     # ── Env-driven auto-configuration ──
     # Optional: read env vars, return a dict of ``PlatformConfig.extra`` fields
     # to seed when the platform is auto-enabled.  Called during
@@ -158,6 +177,19 @@ class PlatformEntry:
     # targets when the gateway is not co-resident with the cron process.
     standalone_sender_fn: Optional[Callable[..., Awaitable[dict]]] = None
 
+    def __post_init__(self) -> None:
+        try:
+            self.agent_tool_policy = AgentToolPolicy(self.agent_tool_policy)
+        except (TypeError, ValueError) as exc:
+            valid = ", ".join(policy.value for policy in AgentToolPolicy)
+            raise ValueError(
+                f"agent_tool_policy must be one of: {valid}"
+            ) from exc
+        if not isinstance(self.inbound_context_references_enabled, bool):
+            raise ValueError(
+                "inbound_context_references_enabled must be a boolean"
+            )
+
 
 class PlatformRegistry:
     """Central registry of platform adapters.

gateway/platforms/base.py

@@ -2190,6 +2190,7 @@ def merge_pending_message_event(
 # reply), an ``EphemeralReply`` to opt the reply into auto-deletion, or
 # ``None`` when the response was already delivered (e.g. via streaming).
 MessageHandler = Callable[[MessageEvent], Awaitable[Optional[Union[str, "EphemeralReply"]]]]
+SessionInterruptHandler = Callable[..., Awaitable[None]]
 
 
 def resolve_channel_prompt(
@@ -2330,6 +2331,11 @@ class BasePlatformAdapter(ABC):
     # set this to False to stay correct-by-default.
     supports_async_delivery: bool = True
 
+    # Stateless request/response adapters may disable the gateway command
+    # surface entirely. When False, dispatch_request leaves plaintext phrases
+    # untouched and rejects slash commands before the gateway handler.
+    request_dispatch_allows_gateway_commands: bool = True
+
     # Whether this adapter's ``send()`` splits long content into multiple
     # messages via ``truncate_message()``.  When True, the delivery router
     # (gateway/delivery.py) skips gateway-level truncation and lets the
@@ -2379,6 +2385,7 @@ def __init__(self, config: PlatformConfig, platform: Platform):
         self.config = config
         self.platform = platform
         self._message_handler: Optional[MessageHandler] = None
+        self._session_interrupt_handler: Optional[SessionInterruptHandler] = None
         # Optional hook (e.g. Telegram DM topic recovery) that rewrites
         # ``event.source.thread_id`` before session keying. Returns the
         # corrected thread_id or None to leave the source untouched.
@@ -2827,6 +2834,57 @@ def set_message_handler(self, handler: MessageHandler) -> None:
         """
         self._message_handler = handler
 
+    async def dispatch_request(
+        self,
+        event: MessageEvent,
+    ) -> Optional[Union[str, "EphemeralReply"]]:
+        """Synchronously dispatch one request and return its final response.
+
+        Stateless request/response transports own task lifecycle, concurrency,
+        delivery, and cancellation. This hook preserves the common inbound
+        preprocessing performed by :meth:`handle_message` while avoiding its
+        background delivery task.
+        """
+        if self._message_handler is None:
+            raise RuntimeError(f"{self.name} message handler is not installed")
+
+        if self.request_dispatch_allows_gateway_commands:
+            coerce_plaintext_gateway_command(event)
+        elif str(getattr(event, "text", "") or "").lstrip().startswith("/"):
+            raise ValueError(f"{self.name} request dispatch does not allow gateway commands")
+        await asyncio.to_thread(self._apply_topic_recovery, event)
+        return await self._message_handler(event)
+
+    def set_session_interrupt_handler(
+        self,
+        handler: Optional[SessionInterruptHandler],
+    ) -> None:
+        """Install the host callback for canceling a running session."""
+        self._session_interrupt_handler = handler
+
+    async def request_session_interrupt(
+        self,
+        source: SessionSource,
+        *,
+        interrupt_reason: str = "Platform requested cancellation",
+        invalidation_reason: str = "platform_cancel",
+    ) -> bool:
+        """Ask the host to interrupt and invalidate a running session.
+
+        Returns ``False`` when the adapter has not been attached to a host.
+        The host callback owns session-key derivation, canonical interruption,
+        and cleanup; callers cannot provide a key or profile namespace.
+        """
+        handler = self._session_interrupt_handler
+        if handler is None:
+            return False
+        await handler(
+            source,
+            interrupt_reason=interrupt_reason,
+            invalidation_reason=invalidation_reason,
+        )
+        return True
+
     def set_topic_recovery_fn(
         self,
         fn: Optional[Callable[[Any], Optional[str]]],
@@ -2932,7 +2990,20 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
         Returns True if connection was successful.
         """
         pass
-
+
+    async def prepare_disconnect(self) -> None:
+        """Quiesce ingress and active work before disconnect authority is revoked.
+
+        Most adapters have no distinct prepare phase. Request/response adapters
+        that must interrupt in-flight sessions during shutdown can override this
+        hook; the gateway calls it while the session interrupt handler is still
+        installed.
+        """
+
+    def active_session_sources(self) -> tuple[SessionSource, ...]:
+        """Snapshot adapter-owned sources that may need canonical interruption."""
+        return ()
+
     @abstractmethod
     async def disconnect(self) -> None:
         """Disconnect from the platform."""