Port AttachmentDownloader and M365AttachmentDownloader from .NET

libraries/microsoft-agents-activity/microsoft_agents/activity/attachment.py

@@ -21,7 +21,7 @@ class Attachment(AgentsModel):
     """
 
     content_type: NonEmptyString
-    content_url: NonEmptyString = None
+    content_url: NonEmptyString | None = None
     content: object = None
-    name: NonEmptyString = None
+    name: NonEmptyString | None = None
     thumbnail_url: NonEmptyString = None

libraries/microsoft-agents-activity/microsoft_agents/activity/channels.py

@@ -79,6 +79,9 @@ class Channels(str, Enum):
     copilot_studio = "pva-studio"
     """Microsoft Copilot Studio channel."""
 
+    m365_copilot = f"msteams:COPILOT"
+    """Microsoft 365 Copilot channel."""
+
     ms_teams = "msteams"
     """Deprecated alias for :attr:`msteams`. Kept for backwards compatibility."""
 

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

@@ -0,0 +1,23 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from email.message import Message
+
+_CONTENT_TYPE = "Content-Type"
+
+
+def _parse_content_type(content_type: str) -> tuple[str, dict[str, str]] | None:
+    """Parses the given content type string into its main type and parameters.
+
+    :param content_type: The content type string to parse.
+    :return: A tuple containing the main content type and a dictionary of parameters,
+             or None if parsing fails.
+    """
+    email = Message()
+    email[_CONTENT_TYPE] = content_type
+    params = email.get_params()
+    if params is None:
+        return None
+    # the first param is the mime-type
+    # the later ones are the attribtues like "charset"
+    return params[0][0], dict(params[1:])

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

@@ -0,0 +1,105 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import json
+from typing import Callable
+
+import aiohttp
+
+from microsoft_agents.activity import (
+    Attachment,
+    Channels,
+    ChannelId,
+)
+
+from microsoft_agents.hosting.core.turn_context import TurnContext
+from microsoft_agents.hosting.core.outbound_host_validator import OutboundHostValidator
+
+from .input_file import InputFileDownloader, InputFile
+from ._utils import _parse_content_type
+
+
+class AttachmentDownloader(InputFileDownloader):
+
+    def __init__(
+        self,
+        client_factory: Callable[[], aiohttp.ClientSession] | None = None,
+        host_validator: OutboundHostValidator | None = None,
+    ):
+        """Constructor for AttachmentDownloader.
+
+        :param client_factory: A callable that returns an aiohttp.ClientSession instance.
+        :param host_validator: An optional OutboundHostValidator instance.
+        """
+
+        self._client_factory = client_factory or aiohttp.ClientSession
+        self._host_validator = host_validator
+
+    async def download_files(self, context: TurnContext) -> list[InputFile]:
+        """Downloads files for the given turn context.
+
+        :param context: The TurnContext instance for the current turn.
+        """
+        if ChannelId.get_channel(context.activity.channel_id) == Channels.ms_teams:
+            return []
+
+        if not context.activity.attachments:
+            return []
+
+        files: list[InputFile] = []
+        for attachment in context.activity.attachments:
+            file = await self._download_file(attachment)
+            if file:
+                files.append(file)
+
+        return files
+
+    async def _download_file(self, attachment: Attachment) -> InputFile | None:
+        """Downloads a single file from the given attachment.
+
+        :param attachment: The attachment to download.
+        :return: An InputFile instance if the download is successful, None otherwise.
+        """
+        if attachment.content_url and (
+            attachment.content_url.startswith("https://")
+            or attachment.content_url.startswith("http://localhost")
+        ):
+            remote_file_url = attachment.content_url
+
+            if (
+                self._host_validator
+                and self._host_validator.enabled
+                and not self._host_validator.is_allowed(remote_file_url)
+            ):
+                return None
+
+            async with self._client_factory() as client:
+                async with client.get(remote_file_url) as response:
+
+                    if not response.status == 200:
+                        return None
+
+                    content_type_val = response.headers.get("Content-Type", "")
+                    result = _parse_content_type(content_type_val)
+                    if result is None:
+                        return None
+                    content_type, _ = result
+                    if content_type.startswith("image/"):
+                        content_type = "image/png"
+
+                    res = await response.read()
+
+                    return InputFile(
+                        content=res,
+                        content_type=content_type,
+                        content_url=attachment.content_url,
+                        filename=attachment.name,
+                    )
+        else:
+            content = bytes(json.dumps(attachment.content), "utf-8")
+            return InputFile(
+                content=content,
+                content_type=attachment.content_type,
+                content_url=attachment.content_url,
+                filename=attachment.name,
+            )

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

@@ -1,13 +1,10 @@
-"""
-Copyright (c) Microsoft Corporation. All rights reserved.
-Licensed under the MIT License.
-"""
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
 
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Optional
 
 from microsoft_agents.hosting.core import TurnContext
 
@@ -26,7 +23,8 @@ class InputFile:
 
     content: bytes
     content_type: str
-    content_url: Optional[str]
+    content_url: str | None = None
+    filename: str | None = None
 
 
 class InputFileDownloader(ABC):

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

@@ -0,0 +1,177 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import json
+from typing import Callable, cast, Any
+
+import aiohttp
+
+from microsoft_agents.activity import (
+    Attachment,
+    Channels,
+    ChannelId,
+)
+
+from microsoft_agents.hosting.core.authorization import (
+    AccessTokenProviderBase,
+    Connections,
+)
+from microsoft_agents.hosting.core.turn_context import TurnContext
+from microsoft_agents.hosting.core.outbound_host_validator import OutboundHostValidator
+
+from .input_file import InputFileDownloader, InputFile
+from ._utils import _parse_content_type
+
+
+class M365AttachmentDownloader(InputFileDownloader):
+    """Downloads attachments from M365/Teams using the configured Token Provider (from Connections)."""
+
+    def __init__(
+        self,
+        connections: Connections,
+        client_factory: Callable[[], aiohttp.ClientSession] | None = None,
+        host_validator: OutboundHostValidator | None = None,
+        *,
+        token_provider_name: str = "",
+        use_anonymous: bool = False,
+        scopes: list[str] | None = None,
+    ):
+        """Constructor for M365AttachmentDownloader.
+
+        :param connections: A Connections instance.
+        :param client_factory: A callable that returns an aiohttp.ClientSession instance.
+        :param host_validator: An optional OutboundHostValidator instance.
+        :param token_provider_name: The name of the token provider.
+        :param use_anonymous: Whether to use anonymous access.
+        :param scopes: A list of scopes for the access token.
+        :param connections: A Connections instance.
+        """
+
+        self._connections = connections
+        self._client_factory = client_factory or aiohttp.ClientSession
+        self._host_validator = host_validator
+
+        self._token_provider_name = token_provider_name
+        self._use_anonymous = use_anonymous
+        self._scopes = scopes or []
+
+    async def download_files(self, context: TurnContext) -> list[InputFile]:
+        """Download files from the given context.
+
+        :param context: The TurnContext instance.
+        :return: A list of InputFile instances.
+        """
+
+        if not context.identity:
+            raise ValueError("No valid context identity found.")
+
+        outgoing_audience_claim = context.identity.get_outgoing_audience_claim()
+        if not outgoing_audience_claim:
+            raise ValueError("No valid outgoing App ID found.")
+
+        if context.activity.channel_id not in (
+            Channels.ms_teams,
+            Channels.m365_copilot,
+        ):
+            return []
+
+        attachments: list[Attachment]
+        if not context.activity.attachments:
+            return []
+        attachments = [
+            att
+            for att in context.activity.attachments
+            if not att.content_type.startswith("text/html")
+        ]
+        if not attachments:
+            return []
+
+        access_token = ""
+
+        if not self._use_anonymous:
+            token_provider: AccessTokenProviderBase | None = None
+            if self._token_provider_name:
+                try:
+                    token_provider = self._connections.get_connection(
+                        self._token_provider_name
+                    )
+                except ValueError:
+                    pass
+            if not token_provider:
+                token_provider = self._connections.get_token_provider_from_activity(
+                    context.identity, context.activity
+                )
+            if not token_provider:
+                raise RuntimeError("No valid token provider found.")
+
+            access_token = await token_provider.get_access_token(
+                outgoing_audience_claim, self._scopes
+            )
+
+        files: list[InputFile] = []
+        for att in attachments:
+            file = await self._download_file(att, access_token)
+            if file:
+                files.append(file)
+
+        return files
+
+    async def _download_file(
+        self, attachment: Attachment, access_token: str
+    ) -> InputFile | None:
+        """Download a single file from the given attachment.
+
+        :param attachment: The Attachment instance.
+        :param access_token: The access token for authentication.
+        :return: An InputFile instance or None if the download fails.
+        """
+        name = attachment.name
+
+        if attachment.content_url and (
+            attachment.content_url.startswith("https://")
+            or attachment.content_url.startswith("http://localhost")
+        ):
+            download_url: str
+            if isinstance(attachment.content, dict):
+                content_dict = cast(dict[str, Any], attachment.content)
+                download_url = content_dict.get("downloadUrl", attachment.content_url)
+            else:
+                download_url = attachment.content_url
+
+            if (
+                self._host_validator is not None
+                and self._host_validator.enabled
+                and not self._host_validator.is_allowed(download_url)
+            ):
+                return None
+
+            async with self._client_factory() as client:
+                async with client.get(
+                    download_url, headers={"Authorization": f"Bearer {access_token}"}
+                ) as response:
+                    if response.status >= 300:
+                        return None
+                    content = await response.read()
+                    result = _parse_content_type(
+                        response.headers.get("Content-Type", "")
+                    )
+                    if result is None:
+                        return None
+                    content_type, _ = result
+                    if content_type.startswith("image/"):
+                        content_type = "image/png"
+
+                    return InputFile(
+                        content=content,
+                        content_type=content_type,
+                        content_url=attachment.content_url,
+                        filename=name,
+                    )
+        else:
+            content = bytes(json.dumps(attachment.content), "utf-8")
+            return InputFile(
+                content=content,
+                content_type=attachment.content_type,
+                content_url=attachment.content_url,
+                filename=attachment.name,
+            )

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/claims_identity.py

@@ -79,6 +79,14 @@ def get_outgoing_app_id(self) -> str | None:
 
         return app_id
 
+    def get_outgoing_audience_claim(self) -> str | None:
+        """Retrieves the audience for an outgoing token from the given incoming activity."""
+        if self.is_agent_claim():
+            return f"api://{self.get_outgoing_app_id()}"
+        if self.is_gov_botframework_claim():
+            return AuthenticationConstants.GOV_AGENTS_SDK_TOKEN_ISSUER
+        return AuthenticationConstants.AGENTS_SDK_SCOPE
+
     def is_agent_claim(self) -> bool:
         """
         Checks if the current claims represents an agent claim (not coming from ABS/SMBA).
@@ -116,6 +124,15 @@ def get_token_audience(self) -> str:
             else AuthenticationConstants.AGENTS_SDK_SCOPE
         )
 
+    def is_gov_botframework_claim(self) -> bool:
+        """Determines whether the specified incoming identity represents a government Bot Framework claim."""
+        aud = self.claims.get(AuthenticationConstants.AUDIENCE_CLAIM, None)
+        return (
+            aud.lower() == AuthenticationConstants.GOV_AGENTS_SDK_TOKEN_ISSUER.lower()
+            if aud
+            else False
+        )
+
     def get_token_scope(self) -> list[str]:
         """
         Gets the token scope from current claims.

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/connector/attachments_base.py

@@ -15,3 +15,7 @@ async def get_attachment_info(self, attachment_id: str) -> AttachmentInfo:
     @abstractmethod
     async def get_attachment(self) -> Optional[AsyncIterator[bytes]]:
         pass
+
+    @abstractmethod
+    def get_attachment_uri(self, attachment_id: str, view_id: str = "original") -> str:
+        pass