Skip to content

feat(teams): migrate outbound send/edit/delete/typing to the SDK (#93 PR 2/4) #144

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
patrick-chinchill merged 1 commit into from
Jun 18, 2026
Merged

feat(teams): migrate outbound send/edit/delete/typing to the SDK (#93 PR 2/4) #144

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
217 changes: 135 additions & 82 deletions src/chat_sdk/adapters/teams/adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -356,8 +356,21 @@ def __init__(self, config: TeamsAdapterConfig | None = None) -> None:
)

self._bot_user_id: str | None = self._app_id or None
# Bot Framework token cache (scope ``api.botframework.com``). Owned by
# ``_get_access_token`` and consumed only by the still-hand-rolled Bot
# Framework paths: native streaming (``_teams_send``, PR 3) and
# ``open_dm``. The SDK ``App`` mints its own Bot Framework token for the
# migrated outbound send/edit/delete/typing paths, so those no longer
# touch this field.
self._access_token: str | None = None
self._token_expiry: float = 0
# Microsoft Graph token cache (scope ``graph.microsoft.com``). Kept on
# DEDICATED fields so it can never collide with the Bot Framework token
# above — the two have different scopes, and sharing one cache caused
# last-writer-wins corruption (issue #93). Owned by ``_get_graph_token``
# and consumed only by the hand-rolled Graph reads.
self._graph_token: str | None = None
self._graph_token_expiry: float = 0
self._token_lock = asyncio.Lock()

# Microsoft Teams SDK ``App`` — owns inbound JWT validation and
Expand Down Expand Up @@ -1327,6 +1340,58 @@ async def _files_to_attachments(self, files: list[FileUpload]) -> list[dict[str,

return attachments

def _point_app_api_at(self, service_url: str) -> None:
"""Aim the SDK ``App``'s Bot Framework client at ``service_url``.

The migrated outbound paths call ``self._app.send(...)`` and
``self._app.api.conversations.activities(...)`` directly (parity with
upstream ``this.app.send`` / ``this.app.api.conversations``). The SDK
binds the App's :class:`ApiClient` to a single service URL at
construction, and ``app.send`` reads ``self.api.service_url`` into the
outgoing :class:`ConversationReference`. Our thread IDs encode a
per-thread service URL, so before each call we retarget the App's API
client — validating against the SSRF allow-list first, exactly as the
retired hand-rolled senders did.

The setter walks the real :class:`ApiClient`'s service-url chain
(the client itself, its ``conversations`` sub-client, and that
sub-client's ``activities_client``). It is defensive about test doubles
that replace ``self._app.api`` with a mock lacking those attributes —
an ``AttributeError`` there is harmless because the mock ignores the
service URL anyway.
"""
_validate_service_url(service_url)
normalized = service_url.rstrip("/")
api = self._app.api
try:
api.service_url = normalized
conversations = api.conversations
conversations.service_url = normalized
conversations.activities_client.service_url = normalized
except AttributeError:
# ``self._app.api`` is a test double without the real client's
# service-url chain; nothing to retarget.
pass
Comment on lines +1343 to +1374

@gemini-code-assist gemini-code-assist Bot Jun 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Critical Race Condition in Concurrent Requests

Since TeamsAdapter is a shared instance (often a singleton) handling concurrent requests, mutating the shared self._app.api service URL in _point_app_api_at introduces a critical race condition.

When multiple concurrent outbound operations (e.g., post_message, edit_message, delete_message, start_typing) are in-flight for different threads with different regional or sovereign service URLs (such as https://smba.trafficmanager.net/amer/ vs https://smba.trafficmanager.net/emea/), they will concurrently overwrite self._app.api.service_url. Because these outbound operations are asynchronous (await self._app.send(...)), one request can yield control and have its service URL overwritten by another concurrent request before or during its API call. This will cause requests to be routed to the wrong regional endpoints, resulting in 404 Not Found or 401 Unauthorized errors.

Suggested Fix

To prevent this, serialize outbound operations that modify the shared API client's service URL using an asyncio.Lock. For example, you can define an async context manager to safely acquire a lock, point the API, and release the lock after the request completes:

# In __init__:
# self._api_lock = asyncio.Lock()

@asynccontextmanager
async def _api_at(self, service_url: str):
    async with self._api_lock:
        self._point_app_api_at(service_url)
        yield

And then wrap the outbound calls:

async with self._api_at(decoded.service_url):
    sent = await self._app.send(...)


@staticmethod
def _message_activity_input(payload: dict[str, Any]) -> Any:
"""Build the SDK ``MessageActivityInput`` from our camelCase activity dict.

The dict we construct (``text`` / ``textFormat`` / ``attachments`` with
``contentType`` / ``contentUrl`` / ``content`` / ``name`` keys) is the
Bot Framework wire shape, which matches the SDK input model's
serialization aliases — so ``model_validate`` round-trips it directly.
We keep building the dict (it is still returned as ``RawMessage.raw``,
preserving the public contract) and convert at the SDK boundary.

Upstream constructs a ``MessageActivity``; the Python SDK splits input
(``MessageActivityInput``) from output (``MessageActivity``) models and
``app.send`` / ``activities.update`` accept only the input variant.
"""
from microsoft_teams.api import MessageActivityInput

return MessageActivityInput.model_validate(payload)

async def post_message(
self,
thread_id: str,
Expand Down Expand Up @@ -1361,8 +1426,16 @@ async def post_message(
)

try:
result = await self._teams_send(decoded, activity_payload)
return RawMessage(id=result.get("id", ""), thread_id=thread_id, raw=activity_payload)
self._point_app_api_at(decoded.service_url)
sent = await self._app.send(
decoded.conversation_id,
self._message_activity_input(activity_payload),
)
return RawMessage(
id=getattr(sent, "id", "") or "",
thread_id=thread_id,
raw=activity_payload,
)
except Exception as error:
self._logger.error(
"Teams API: send failed",
Expand All @@ -1371,10 +1444,7 @@ async def post_message(
"error": str(error),
},
)
error_dict: dict[str, Any] = {"message": str(error)}
if hasattr(error, "status"):
error_dict["statusCode"] = error.status
_handle_teams_error(error_dict, "postMessage")
_handle_teams_error(error, "postMessage")
raise # unreachable: _handle_teams_error always raises

# Regular text message
Expand All @@ -1401,8 +1471,17 @@ async def post_message(
)

try:
result = await self._teams_send(decoded, activity_payload)
return RawMessage(id=result.get("id", ""), thread_id=thread_id, raw=activity_payload)
self._point_app_api_at(decoded.service_url)
sent = await self._app.send(
decoded.conversation_id,
self._message_activity_input(activity_payload),
)
self._logger.debug("Teams API: send response", {"messageId": getattr(sent, "id", None)})
return RawMessage(
id=getattr(sent, "id", "") or "",
thread_id=thread_id,
raw=activity_payload,
)
except Exception as error:
self._logger.error(
"Teams API: send failed",
Expand All @@ -1411,10 +1490,7 @@ async def post_message(
"error": str(error),
},
)
error_dict = {"message": str(error)}
if hasattr(error, "status"):
error_dict["statusCode"] = error.status
_handle_teams_error(error_dict, "postMessage")
_handle_teams_error(error, "postMessage")

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve SDK HTTP statuses when mapping errors

When the SDK call fails, this now passes the raw exception into _handle_teams_error, but the Microsoft Teams SDK HTTP client raises httpx.HTTPStatusError via response.raise_for_status(), where the status is on error.response.status_code rather than direct status_code/status/code attributes. In real 401/403/429 responses from app.send (and the same pattern in edit/delete), the mapper therefore falls through to a generic NetworkError instead of the existing AuthenticationError, permission, or rate-limit errors, so callers lose the actionable failure type.

Useful? React with 👍 / 👎.

# Should not reach here due to _handle_teams_error always raising
raise # pragma: no cover

Expand Down Expand Up @@ -1467,7 +1543,11 @@ async def edit_message(
)

try:
await self._teams_update(decoded, message_id, activity_payload)
self._point_app_api_at(decoded.service_url)
await self._app.api.conversations.activities(decoded.conversation_id).update(
message_id,
self._message_activity_input(activity_payload),
)
except Exception as error:
self._logger.error(
"Teams API: updateActivity failed",
Expand All @@ -1477,10 +1557,7 @@ async def edit_message(
"error": str(error),
},
)
error_dict = {"message": str(error)}
if hasattr(error, "status"):
error_dict["statusCode"] = error.status
_handle_teams_error(error_dict, "editMessage")
_handle_teams_error(error, "editMessage")
raise # unreachable: _handle_teams_error always raises

return RawMessage(id=message_id, thread_id=thread_id, raw=activity_payload)
Expand All @@ -1498,7 +1575,8 @@ async def delete_message(self, thread_id: str, message_id: str) -> None:
)

try:
await self._teams_delete(decoded, message_id)
self._point_app_api_at(decoded.service_url)
await self._app.api.conversations.activities(decoded.conversation_id).delete(message_id)
except Exception as error:
self._logger.error(
"Teams API: deleteActivity failed",
Expand All @@ -1508,10 +1586,7 @@ async def delete_message(self, thread_id: str, message_id: str) -> None:
"error": str(error),
},
)
error_dict = {"message": str(error)}
if hasattr(error, "status"):
error_dict["statusCode"] = error.status
_handle_teams_error(error_dict, "deleteMessage")
_handle_teams_error(error, "deleteMessage")
raise # unreachable: _handle_teams_error always raises

async def add_reaction(
Expand All @@ -1534,6 +1609,8 @@ async def remove_reaction(

async def start_typing(self, thread_id: str, status: str | None = None) -> None:
"""Send typing indicator to a Teams conversation."""
from microsoft_teams.api import TypingActivityInput

decoded = self.decode_thread_id(thread_id)

self._logger.debug(
Expand All @@ -1544,7 +1621,8 @@ async def start_typing(self, thread_id: str, status: str | None = None) -> None:
)

try:
await self._teams_send(decoded, {"type": "typing"})
self._point_app_api_at(decoded.service_url)
await self._app.send(decoded.conversation_id, TypingActivityInput())
except Exception as error:
self._logger.error(
"Teams API: send (typing) failed",
Expand Down Expand Up @@ -2751,17 +2829,25 @@ def _extract_attachments_from_graph_message(self, msg: dict[str, Any]) -> list[A
return attachments

async def _get_graph_token(self) -> str:
"""Get a Microsoft Graph API access token (OAuth2 client credentials)."""
"""Get a Microsoft Graph API access token (OAuth2 client credentials).

Caches on the DEDICATED ``_graph_token`` / ``_graph_token_expiry``
fields — never the Bot Framework ``_access_token`` field. The two
tokens carry different scopes (``graph.microsoft.com`` vs
``api.botframework.com``); sharing one cache slot let whichever was
fetched last clobber the other, so a Graph read could end up sending a
Bot Framework token (and vice versa). See issue #93.
"""
import time as _time

# Reuse cached token if valid
if self._access_token and _time.time() < self._token_expiry:
return self._access_token
if self._graph_token and _time.time() < self._graph_token_expiry:
return self._graph_token

async with self._token_lock:
# Double-check after acquiring lock to avoid redundant refreshes
if self._access_token and _time.time() < self._token_expiry:
return self._access_token
if self._graph_token and _time.time() < self._graph_token_expiry:
return self._graph_token

tenant_id = self._app_tenant_id or "botframework.com"
token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
Expand All @@ -2783,16 +2869,25 @@ async def _get_graph_token(self) -> str:
f"Failed to get Graph API token: {response.status} {error_text}",
)
data = await response.json()
self._access_token = data["access_token"]
self._token_expiry = _time.time() + data.get("expires_in", 3600) - 300
return self._access_token # type: ignore[return-value]
self._graph_token = data["access_token"]
self._graph_token_expiry = _time.time() + data.get("expires_in", 3600) - 300
return self._graph_token # type: ignore[return-value]

# =========================================================================
# Teams Bot Framework HTTP API helpers
# =========================================================================

async def _get_access_token(self) -> str:
"""Get a Bot Framework access token (OAuth2 client credentials)."""
"""Get a Bot Framework access token (OAuth2 client credentials).

Scope ``api.botframework.com``, cached on ``_access_token`` /
``_token_expiry``. The migrated outbound paths
(post/edit/delete/typing) now mint their Bot Framework token through
the SDK ``App``, so this hand-rolled token is consumed only by the
still-hand-rolled Bot Framework callers: native streaming via
:meth:`_teams_send` (PR 3) and :meth:`open_dm`. It must never share a
cache slot with the Graph token (see :meth:`_get_graph_token`).
"""
import time

if self._access_token and time.time() < self._token_expiry:
Expand Down Expand Up @@ -2843,7 +2938,14 @@ async def _teams_send(
decoded: TeamsThreadId,
activity: dict[str, Any],
) -> dict[str, Any]:
"""Send an activity to a Teams conversation via Bot Framework REST API."""
"""Send an activity to a Teams conversation via Bot Framework REST API.

Retained for the still-hand-rolled native streaming path (PR 3), which
needs the raw ``channelData``/``entities`` streaming envelope and the
server-assigned ``streamId`` from the REST response — neither of which
the SDK ``app.send`` surface exposes today. The migrated outbound
public methods no longer use this; they delegate to the SDK.
"""
_validate_service_url(decoded.service_url)
token = await self._get_access_token()
url = f"{decoded.service_url}v3/conversations/{decoded.conversation_id}/activities"
Expand All @@ -2865,55 +2967,6 @@ async def _teams_send(
)
return await response.json()

async def _teams_update(
self,
decoded: TeamsThreadId,
message_id: str,
activity: dict[str, Any],
) -> None:
"""Update an activity in a Teams conversation via Bot Framework REST API."""
_validate_service_url(decoded.service_url)
token = await self._get_access_token()
url = f"{decoded.service_url}v3/conversations/{decoded.conversation_id}/activities/{message_id}"

session = await self._get_http_session()
async with session.put(
url,
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
},
json=activity,
) as response:
if not response.ok:
error_text = await response.text()
raise NetworkError(
"teams",
f"Teams API error: {response.status} {error_text}",
)

async def _teams_delete(
self,
decoded: TeamsThreadId,
message_id: str,
) -> None:
"""Delete an activity from a Teams conversation via Bot Framework REST API."""
_validate_service_url(decoded.service_url)
token = await self._get_access_token()
url = f"{decoded.service_url}v3/conversations/{decoded.conversation_id}/activities/{message_id}"

session = await self._get_http_session()
async with session.delete(
url,
headers={"Authorization": f"Bearer {token}"},
) as response:
if not response.ok:
error_text = await response.text()
raise NetworkError(
"teams",
f"Teams API error: {response.status} {error_text}",
)


def create_teams_adapter(config: TeamsAdapterConfig | None = None) -> TeamsAdapter:
"""Factory function to create a Teams adapter."""
Expand Down
Loading
Loading