Agentic header propagation

dev/integration/tests/agentic/__init__.py

@@ -0,0 +1,2 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.

dev/integration/tests/agentic/test_agentic_header_propagation.py

@@ -0,0 +1,157 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+"""Integration tests for agentic header propagation."""
+
+import asyncio
+
+import pytest
+from aiohttp import ClientSession, web
+from aiohttp.test_utils import TestServer
+
+from microsoft_agents.activity import Activity, ActivityTypes, RoleTypes
+from microsoft_agents.hosting.aiohttp import CloudAdapter, start_agent_process
+from microsoft_agents.hosting.core import (
+    AgentApplication,
+    AgentAuthConfiguration,
+    ApplicationOptions,
+    Authorization,
+    ConnectorClientBase,
+    MemoryStorage,
+    TurnContext,
+    TurnState,
+)
+from microsoft_agents.hosting.core.authorization import ClaimsIdentity
+
+
+class _FakeTokenProvider:
+    def __init__(self):
+        self._configuration = AgentAuthConfiguration()
+
+    @property
+    def configuration(self) -> AgentAuthConfiguration:
+        return self._configuration
+
+    async def get_access_token(
+        self, resource_url: str, scopes: list[str], force_refresh: bool = False
+    ) -> str:
+        return "test-access-token"
+
+    async def get_agentic_user_token(
+        self,
+        tenant_id: str,
+        agent_app_instance_id: str,
+        agentic_user_id: str,
+        scopes: list[str],
+    ) -> str:
+        return "test-agentic-user-token"
+
+
+class _FakeConnections:
+    def __init__(self):
+        self._provider = _FakeTokenProvider()
+
+    def get_connection(self, connection_name: str):
+        return self._provider
+
+    def get_default_connection(self):
+        return self._provider
+
+    def get_token_provider(self, claims_identity: ClaimsIdentity, service_url: str):
+        return self._provider
+
+    def get_token_provider_from_activity(
+        self, claims_identity: ClaimsIdentity, activity: Activity
+    ):
+        return self._provider
+
+    def get_default_connection_configuration(self) -> AgentAuthConfiguration:
+        return self._provider.configuration
+
+
+@pytest.mark.asyncio
+async def test_agentic_turn_propagates_headers_on_connector_client_request():
+    captured_headers = {}
+    callback_received = asyncio.Event()
+
+    async def callback_handler(request: web.Request) -> web.Response:
+        captured_headers.update(dict(request.headers))
+        callback_received.set()
+        return web.json_response({"id": "connector-response-id"})
+
+    callback_app = web.Application()
+    callback_app.router.add_post("/v3/conversations/{tail:.*}", callback_handler)
+    callback_server = TestServer(callback_app)
+    await callback_server.start_server()
+
+    connection_manager = _FakeConnections()
+    storage = MemoryStorage()
+    adapter = CloudAdapter(connection_manager=connection_manager)
+    agent_application = AgentApplication[TurnState](
+        options=ApplicationOptions(
+            storage=storage,
+            start_typing_timer=False,
+            remove_recipient_mention=False,
+        ),
+        authorization=Authorization(storage, connection_manager),
+        agent_name="Agentic Header Test Agent",
+    )
+
+    @agent_application.activity(ActivityTypes.message)
+    async def on_message(context: TurnContext, state: TurnState) -> None:
+        connector_client = context.services.get(ConnectorClientBase)
+        assert connector_client is not None
+
+        await connector_client.conversations.send_to_conversation(
+            context.activity.conversation.id,
+            Activity(type=ActivityTypes.message, text="connector client response"),
+        )
+
+    agent_app = web.Application()
+
+    async def messages(request: web.Request) -> web.Response:
+        return await start_agent_process(
+            request,
+            agent_application=agent_application,
+            adapter=adapter,
+        )
+
+    agent_app.router.add_post("/api/messages", messages)
+    agent_server = TestServer(agent_app)
+    await agent_server.start_server()
+
+    try:
+        activity = Activity(
+            type=ActivityTypes.message,
+            text="send with connector client",
+            channel_id="msteams:Copilot",
+            service_url=str(callback_server.make_url("/")),
+            conversation={"id": "conversation-id"},
+            from_property={"id": "user-id", "role": RoleTypes.user},
+            recipient={
+                "id": "agent-id",
+                "role": RoleTypes.agentic_user,
+                "agentic_app_id": "Entra:agentic-app-id",
+                "agentic_user_id": "agentic-user-id",
+                "tenant_id": "tenant-id",
+            },
+        )
+
+        async with ClientSession() as session:
+            async with session.post(
+                agent_server.make_url("/api/messages"),
+                json=activity.model_dump(
+                    by_alias=True, exclude_unset=True, exclude_none=True, mode="json"
+                ),
+            ) as response:
+                assert response.status == 202
+
+        await asyncio.wait_for(callback_received.wait(), timeout=5)
+
+        assert captured_headers["AgentRegistrar"] == "A365"
+        assert captured_headers["AgentID"] == "Entra:agentic-app-id"
+        assert captured_headers["AgentName"] == "Agentic Header Test Agent"
+        assert captured_headers["Agent-Referrer"] == "msteams:Copilot"
+    finally:
+        await agent_server.close()
+        await callback_server.close()

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/__init__.py

@@ -89,6 +89,13 @@
     get_product_info,
 )
 
+# Header propagation
+from .header_propagation import (
+    HeaderValueProvider,
+    AgenticHeaderProvider,
+    HeaderPropagationContext,
+)
+
 # State management
 from .state.agent_state import AgentState
 from .state.state_property_accessor import StatePropertyAccessor
@@ -168,6 +175,9 @@
     "TeamsConnectorClient",
     "ConnectorClientBase",
     "get_product_info",
+    "HeaderValueProvider",
+    "AgenticHeaderProvider",
+    "HeaderPropagationContext",
     "AgentState",
     "StatePropertyAccessor",
     "UserState",

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py

@@ -36,6 +36,7 @@
 
 from ..agent import Agent
 from ..authorization import Connections
+from ..header_propagation import AgenticHeaderProvider, HeaderPropagationContext
 from .app_error import ApplicationError
 from .app_options import ApplicationOptions
 
@@ -108,6 +109,14 @@ def __init__(
         self._internal_before_turn = []
         self._internal_after_turn = []
 
+        # Human-friendly agent name surfaced on outgoing agentic headers.
+        # Falls back to the application class name when not explicitly provided.
+        raw_agent_name = kwargs.get("agent_name") or type(self).__name__
+        sanitized_agent_name = re.sub(
+            r"[^A-Za-z0-9 ._-]", "", str(raw_agent_name)
+        ).strip()
+        self._agent_name = sanitized_agent_name or type(self).__name__
+
         configuration = kwargs
 
         if not options:
@@ -184,7 +193,8 @@ def __init__(
             auth_options = {
                 key: value
                 for key, value in configuration.items()
-                if key not in ["storage", "connection_manager", "handlers"]
+                if key
+                not in ["storage", "connection_manager", "handlers", "agent_name"]
             }
             self._auth = Authorization(
                 storage=self._storage,
@@ -810,6 +820,15 @@ async def on_turn(self, context: TurnContext):
 
     async def _on_turn(self, context: TurnContext):
         try:
+            # Register Activity-derived header provider for agentic requests so
+            # that agent identity headers are propagated on outgoing requests
+            # made while processing this turn.
+            HeaderPropagationContext.reset()
+            if context.activity and context.activity.is_agentic_request():
+                HeaderPropagationContext.register(
+                    AgenticHeaderProvider(context.activity, self._agent_name)
+                )
+
             with spans.AppOnTurn(context) as on_turn_span:
                 use_typing = (
                     self._options.start_typing_timer

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/channel_service_adapter.py

@@ -5,7 +5,7 @@
 
 from abc import ABC
 from http import HTTPStatus
-from typing import Awaitable, Callable, Optional
+from typing import Awaitable, Callable, Optional, cast
 from uuid import uuid4
 
 from microsoft_agents.activity import (
@@ -26,9 +26,7 @@
 )
 from microsoft_agents.hosting.core.connector import (
     ConnectorClientBase,
-    ConnectorClient,
     UserTokenClientBase,
-    UserTokenClient,
 )
 from microsoft_agents.hosting.core.authorization import (
     AuthenticationConstants,

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/connector/client/__init__.py

@@ -1,3 +1,6 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
 from .connector_client import ConnectorClient
 from .user_token_client import UserTokenClient
 

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/connector/client/_base_client.py

@@ -0,0 +1,94 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import logging
+from typing import Any, Callable
+
+from aiohttp import ClientSession
+
+from ...header_propagation import HeaderPropagationContext
+
+logger = logging.getLogger(__name__)
+
+
+class _ClientSessionWrapper:
+    """ClientSession wrapper that merges propagated headers per request."""
+
+    def __init__(self, session: ClientSession):
+        self._session = session
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._session, name)
+
+    def _separate_headers(self, **kwargs) -> tuple[dict, dict]:
+        """
+        Separate headers from other keyword arguments.
+
+        :param kwargs: Keyword arguments that may contain headers.
+        :return: A tuple containing the headers and the remaining keyword arguments.
+        """
+        headers = dict(kwargs.get("headers") or {})
+        kwargs_without_headers = {k: v for k, v in kwargs.items() if k != "headers"}
+        return headers, kwargs_without_headers
+
+    def _apply_headers(self, headers: dict) -> None:
+        """
+        Merge propagated headers into the request headers.
+
+        Explicit request headers take precedence over propagated values with the
+        same name.
+
+        :param headers: Mutable request headers to augment.
+        """
+        propagated_headers = HeaderPropagationContext.collect_headers()
+        if propagated_headers:
+            for key, value in propagated_headers.items():
+                headers.setdefault(key, value)
+            logger.debug(
+                "Applying propagated headers: %s", list(propagated_headers.keys())
+            )
+
+    def _call_with_headers(self, method: Callable, *args, **kwargs):
+        """
+        Call the underlying session method with propagated headers merged.
+
+        :param method: The HTTP method to call.
+        :param args: Positional arguments for the method.
+        :param kwargs: Keyword arguments for the method.
+        :return: The result of the method call.
+        """
+        headers, kwargs_without_headers = self._separate_headers(**kwargs)
+        self._apply_headers(headers)
+        return method(*args, headers=headers, **kwargs_without_headers)
+
+    def request(self, *args, **kwargs):
+        return self._call_with_headers(self._session.request, *args, **kwargs)
+
+    def get(self, *args, **kwargs):
+        return self._call_with_headers(self._session.get, *args, **kwargs)
+
+    def post(self, *args, **kwargs):
+        return self._call_with_headers(self._session.post, *args, **kwargs)
+
+    def put(self, *args, **kwargs):
+        return self._call_with_headers(self._session.put, *args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        return self._call_with_headers(self._session.delete, *args, **kwargs)
+
+    def patch(self, *args, **kwargs):
+        return self._call_with_headers(self._session.patch, *args, **kwargs)
+
+
+class _BaseClient:
+
+    def __init__(self, client: ClientSession):
+        self._client = client
+
+    def _wrapped_client(self) -> _ClientSessionWrapper:
+        """
+        Returns a session wrapper that merges propagated headers per request.
+
+        :return: The wrapped ClientSession.
+        """
+        return _ClientSessionWrapper(self._client)