botas

 90class Attachment(_CamelModel):
 91    """Bot Service attachment (images, cards, files, etc.).
 92
 93    Attributes:
 94        content_type: MIME type (e.g. ``"application/vnd.microsoft.card.adaptive"``).
 95        content_url: URL to download the attachment content.
 96        content: Inline attachment content (e.g. an Adaptive Card JSON object).
 97        name: Display name / filename.
 98        thumbnail_url: URL to a thumbnail image.
 99    """
100
101    content_type: str
102    content_url: Optional[str] = None
103    content: Any = None
104    name: Optional[str] = None
105    thumbnail_url: Optional[str] = None

117class BotApplication:
118    """Central entry point for building a bot with the Bot Service.
119
120    Manages the middleware pipeline, activity handler dispatch, outbound
121    messaging via :class:`ConversationClient`, and OAuth2 token lifecycle
122    via :class:`TokenManager`.
123
124    Supports async context-manager usage for automatic resource cleanup::
125
126        async with BotApplication(options) as bot:
127            bot.on("message", my_handler)
128            ...
129
130    Attributes:
131        version: Library version string.
132        conversation_client: Client for sending outbound activities.
133        on_activity: Optional catch-all handler invoked for every activity type.
134    """
135
136    version: str = __import__("botas._version", fromlist=["__version__"]).__version__
137
138    def __init__(self, options: BotApplicationOptions = BotApplicationOptions()) -> None:
139        """Initialise the bot application.
140
141        Args:
142            options: Configuration for authentication credentials and token
143                acquisition.  Defaults to reading from environment variables.
144        """
145        self._token_manager = TokenManager(options)
146        token_provider = self._token_manager.get_bot_token
147        self.conversation_client = ConversationClient(token_provider)
148        self._middlewares: list[TurnMiddleware] = []
149        self._handlers: dict[str, _ActivityHandler] = {}
150        self._invoke_handlers: dict[str, _InvokeActivityHandler] = {}
151        self.on_activity: Optional[_ActivityHandler] = None
152
153    @property
154    def appid(self) -> Optional[str]:
155        """The bot application/client ID exposed from the token manager."""
156        return self._token_manager.client_id
157
158    def on(
159        self,
160        type: str,
161        handler: Optional[_ActivityHandler] = None,
162    ) -> Any:
163        """Register a handler for an activity type.
164
165        Only one handler is stored per type; re-registering the same type
166        replaces the previous handler.
167
168        Can be used as a two-argument call or as a decorator::
169
170            bot.on('message', my_handler)
171
172            @bot.on('message')
173            async def my_handler(ctx: TurnContext):
174                await ctx.send("hello")
175
176        Args:
177            type: The activity type to handle (e.g. ``"message"``, ``"typing"``).
178            handler: Async handler function.  If omitted, returns a decorator.
179
180        Returns:
181            The ``BotApplication`` instance when called with a handler, or a
182            decorator function when called without one.
183        """
184        if handler is None:
185
186            def decorator(fn: _ActivityHandler) -> _ActivityHandler:
187                self._handlers[type.lower()] = fn
188                return fn
189
190            return decorator
191        self._handlers[type.lower()] = handler
192        return self
193
194    def use(self, middleware: TurnMiddleware) -> "BotApplication":
195        """Register a middleware in the turn pipeline.
196
197        Middleware executes in registration order before handler dispatch.
198        Each middleware receives ``(context, next)`` and must call ``next()``
199        to continue the pipeline, or skip it to short-circuit processing.
200
201        Args:
202            middleware: An object implementing :class:`TurnMiddleware`.
203
204        Returns:
205            The ``BotApplication`` instance for chaining.
206        """
207        self._middlewares.append(middleware)
208        return self
209
210    def use_state(self, storage: Any) -> "BotApplication":
211        """Enable turn state management with the given storage backend.
212
213        Registers state middleware that loads state at turn start and saves
214        dirty state at turn end (only if the turn completes successfully).
215
216        Args:
217            storage: A storage implementation (MemoryStorage, FileStorage, etc.).
218
219        Returns:
220            The ``BotApplication`` instance for chaining.
221
222        Example::
223
224            from botas.state import MemoryStorage
225
226            bot = BotApplication()
227            bot.use_state(MemoryStorage())
228        """
229        import asyncio
230
231        from botas.state import TurnState
232
233        # Per (conversation_key, user_key) lock so concurrent turns for the SAME
234        # user/conversation serialize their load -> handler -> save sequence and
235        # avoid lost updates (#365). Different users/conversations do not block
236        # each other. Locks are created lazily under `pair_locks_guard`.
237        pair_locks: dict[tuple[str, str], asyncio.Lock] = {}
238        pair_locks_guard = asyncio.Lock()
239
240        async def get_pair_lock(key_pair: tuple[str, str]) -> asyncio.Lock:
241            async with pair_locks_guard:
242                lock = pair_locks.get(key_pair)
243                if lock is None:
244                    lock = asyncio.Lock()
245                    pair_locks[key_pair] = lock
246                return lock
247
248        async def state_middleware(context: TurnContext, next: Callable[[], Awaitable[None]]) -> None:
249            # Load state at turn start
250            conversation_key = TurnState.derive_conversation_key(context.activity)
251            user_key = TurnState.derive_user_key(context.activity)
252            keys = [conversation_key, user_key]
253
254            # Acquire per (conv, user) lock so the load -> handler -> save sequence
255            # is atomic against concurrent turns for the same key pair.
256            pair_lock = await get_pair_lock((conversation_key, user_key))
257            async with pair_lock:
258                loaded = await storage.read(keys)
259                conversation_data = loaded.get(conversation_key)
260                user_data = loaded.get(user_key)
261
262                # Initialize TurnState and attach to context
263                context.state = TurnState(
264                    context.activity,
265                    conversation_data,  # type: ignore
266                    user_data,  # type: ignore
267                )
268
269                # Call next and save state ONLY if no exception
270                exception_raised = False
271                try:
272                    await next()
273                except Exception:
274                    exception_raised = True
275                    raise
276                finally:
277                    if not exception_raised:
278                        # Save dirty state
279                        changes = {}
280                        deletions = []
281
282                        if context.state.conversation.is_deleted():
283                            deletions.append(conversation_key)
284                        elif context.state.conversation.is_dirty():
285                            changes[conversation_key] = context.state.conversation.to_dict()
286
287                        if context.state.user.is_deleted():
288                            deletions.append(user_key)
289                        elif context.state.user.is_dirty():
290                            changes[user_key] = context.state.user.to_dict()
291
292                        if changes:
293                            await storage.write(changes)
294                        if deletions:
295                            await storage.delete(deletions)
296
297        # Create a middleware object from the function
298        class StateMiddleware:
299            async def on_turn(self, context: TurnContext, next: Callable[[], Awaitable[None]]) -> None:  # noqa: A003
300                await state_middleware(context, next)
301
302        self._middlewares.append(StateMiddleware())
303        return self
304
305    def on_invoke(
306        self,
307        name: str,
308        handler: Optional[_InvokeActivityHandler] = None,
309    ) -> Any:
310        """Register a handler for an invoke activity by its ``activity.name`` sub-type.
311
312        The handler must return an :class:`InvokeResponse`.  Only one handler
313        per name is supported; re-registering the same name replaces the
314        previous handler.
315
316        Can be used as a two-argument call or as a decorator::
317
318            bot.on_invoke("adaptiveCard/action", my_handler)
319
320            @bot.on_invoke("adaptiveCard/action")
321            async def my_handler(ctx): ...
322        """
323        if handler is None:
324
325            def decorator(fn: _InvokeActivityHandler) -> _InvokeActivityHandler:
326                self._invoke_handlers[name.lower()] = fn
327                return fn
328
329            return decorator
330        self._invoke_handlers[name.lower()] = handler
331        return self
332
333    async def process_body(self, body: str) -> Optional[InvokeResponse]:
334        """Parse and process a raw JSON activity body.
335
336        Deserializes the JSON string into a :class:`CoreActivity`, validates
337        required fields and the ``serviceUrl``, then runs the full middleware
338        pipeline followed by handler dispatch.
339
340        For ``invoke`` activities, returns the :class:`InvokeResponse` produced
341        by the registered handler, a 200 response if no invoke handlers are
342        registered, or a 501 response if handlers exist but none match.
343        Returns ``None`` for all other activity types.
344
345        Args:
346            body: Raw JSON string representing a Bot Service activity.
347
348        Returns:
349            An :class:`InvokeResponse` for invoke activities, or ``None``.
350
351        Raises:
352            ValueError: If the JSON is malformed or required activity fields
353                are missing.
354            BotHandlerException: If the matched handler raises an exception.
355        """
356        try:
357            activity = CoreActivity.model_validate_json(body)
358        except json.JSONDecodeError as exc:
359            raise ValueError("Invalid JSON in request body") from exc
360        _assert_activity(activity)
361        _validate_service_url(activity.service_url)
362
363        metrics = get_metrics()
364        if metrics:
365            metrics.activities_received.add(1, {"activity.type": activity.type or ""})
366        start_time = time.perf_counter()
367
368        with _span(
369            "botas.turn",
370            **{
371                "activity.type": activity.type or "",
372                "activity.id": activity.id or "",
373                "conversation.id": activity.conversation.id if activity.conversation else "",
374                "channel.id": activity.channel_id or "",
375                "bot.id": self._token_manager.client_id or "",
376            },
377        ):
378            try:
379                return await self._run_pipeline(activity)
380            finally:
381                if metrics:
382                    elapsed_ms = (time.perf_counter() - start_time) * 1000
383                    metrics.turn_duration.record(elapsed_ms, {"activity.type": activity.type or ""})
384
385    async def send_activity_async(
386        self,
387        service_url: str,
388        conversation_id: str,
389        activity: Union[
390            CoreActivity,
391            dict[str, Any],
392        ],
393    ) -> Optional[ResourceResponse]:
394        """Proactively send an activity to a conversation.
395
396        Use this to push messages outside of the normal turn pipeline (e.g.
397        notifications or proactive messages).
398
399        Args:
400            service_url: The channel's service URL.
401            conversation_id: Target conversation identifier.
402            activity: The activity payload to send.
403
404        Returns:
405            A :class:`ResourceResponse` with the new activity ID, or ``None``
406            if the channel does not return one.
407        """
408        return await self.conversation_client.send_activity_async(service_url, conversation_id, activity)
409
410    async def aclose(self) -> None:
411        """Close the underlying HTTP client and release resources.
412
413        Should be called during application shutdown.  Alternatively, use the
414        bot as an async context manager to ensure automatic cleanup.
415        """
416        await self.conversation_client.aclose()
417
418    async def __aenter__(self) -> "BotApplication":
419        """Enter the async context manager."""
420        return self
421
422    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
423        """Exit the async context manager, ensuring resources are closed."""
424        await self.aclose()
425
426    async def _handle_activity_async(self, context: TurnContext) -> Optional[InvokeResponse]:
427        if context.activity.type == "invoke":
428            return await self._dispatch_invoke_async(context)
429        handler = self.on_activity or self._handlers.get(context.activity.type.lower())
430        if handler is None:
431            return None
432        dispatch_mode = "on_activity" if self.on_activity else "typed"
433        with _span(
434            "botas.handler",
435            **{"handler.type": context.activity.type, "handler.dispatch": dispatch_mode},
436        ):
437            try:
438                await handler(context)
439            except Exception as exc:
440                metrics = get_metrics()
441                if metrics:
442                    metrics.handler_errors.add(1, {"activity.type": context.activity.type})
443                raise BotHandlerException(
444                    f'Handler for "{context.activity.type}" threw an error',
445                    exc,
446                    context.activity,
447                ) from exc
448        return None
449
450    async def _dispatch_invoke_async(self, context: TurnContext) -> InvokeResponse:
451        if not self._invoke_handlers:
452            return InvokeResponse(status=200, body={})
453        name = context.activity.name
454        handler = self._invoke_handlers.get(name.lower()) if name else None
455        if handler is None:
456            return InvokeResponse(status=501)
457        with _span(
458            "botas.handler",
459            **{"handler.type": "invoke", "handler.dispatch": "invoke"},
460        ):
461            try:
462                return await handler(context)
463            except Exception as exc:
464                metrics = get_metrics()
465                if metrics:
466                    metrics.handler_errors.add(1, {"activity.type": "invoke"})
467                raise BotHandlerException(
468                    f'Invoke handler for "{name}" threw an error',
469                    exc,
470                    context.activity,
471                ) from exc
472
473    async def _run_pipeline(self, activity: CoreActivity) -> Optional[InvokeResponse]:
474        context = TurnContext(self, activity)
475        index = 0
476        invoke_response: Optional[InvokeResponse] = None
477
478        async def next_fn() -> None:
479            nonlocal index, invoke_response
480            if index < len(self._middlewares):
481                mw = self._middlewares[index]
482                mw_index = index
483                index += 1
484                mw_name = type(mw).__name__
485                mw_start = time.perf_counter()
486                with _span(
487                    "botas.middleware",
488                    **{"middleware.name": mw_name, "middleware.index": mw_index},
489                ):
490                    try:
491                        await mw.on_turn(context, next_fn)
492                    finally:
493                        metrics = get_metrics()
494                        if metrics:
495                            elapsed_ms = (time.perf_counter() - mw_start) * 1000
496                            metrics.middleware_duration.record(elapsed_ms, {"middleware.name": mw_name})
497            else:
498                invoke_response = await self._handle_activity_async(context)
499
500        await next_fn()
501        return invoke_response

22@dataclass
23class BotApplicationOptions:
24    """Configuration options for :class:`BotApplication` authentication.
25
26    All fields are optional; when ``None``, values are read from environment
27    variables (``CLIENT_ID``, ``CLIENT_SECRET``, ``TENANT_ID``,
28    ``MANAGED_IDENTITY_CLIENT_ID``).
29
30    Attributes:
31        client_id: Azure AD application (bot) ID.
32        client_secret: Azure AD client secret.
33        tenant_id: Azure AD tenant ID (defaults to ``"common"``).
34        managed_identity_client_id: Client ID for managed identity auth.
35        token_factory: Custom async callable ``(scope, tenant) -> token``
36            that bypasses MSAL entirely.
37    """
38
39    client_id: Optional[str] = None
40    client_secret: Optional[str] = None
41    tenant_id: Optional[str] = None
42    managed_identity_client_id: Optional[str] = None
43    token_factory: Optional[Callable[[str, str], Awaitable[str]]] = None

36class BotAuthError(Exception):
37    """Raised when inbound JWT validation fails.
38
39    Inspect the message for the specific reason (expired, bad audience, etc.).
40    """
41
42    pass

 89class BotHandlerException(Exception):
 90    """Wraps an exception thrown inside an activity handler.
 91
 92    When an activity handler or invoke handler raises, the exception is
 93    caught by the pipeline and re-raised as a ``BotHandlerException`` with
 94    the original exception attached as ``cause`` and ``__cause__``.
 95
 96    Attributes:
 97        name: Always ``"BotHandlerException"``.
 98        cause: The original exception raised by the handler.
 99        activity: The activity being processed when the error occurred.
100    """
101
102    def __init__(self, message: str, cause: BaseException, activity: CoreActivity) -> None:
103        """Initialise a BotHandlerException.
104
105        Args:
106            message: Human-readable description of the failure.
107            cause: The original exception raised by the handler.
108            activity: The activity that was being processed.
109        """
110        super().__init__(message)
111        self.name = "BotHandlerException"
112        self.cause = cause
113        self.activity = activity
114        self.__cause__ = cause

20class CardAction(_CamelModel):
21    """A clickable action button presented to the user.
22
23    Attributes:
24        type: Action type (``"imBack"``, ``"postBack"``, ``"openUrl"``, etc.).
25        title: Button label displayed to the user.
26        value: Value sent back to the bot when the button is clicked.
27        text: Text sent to the bot (for ``imBack`` actions).
28        display_text: Text displayed in the chat when the button is clicked.
29        image: URL of an icon image for the button.
30    """
31
32    type: str = "imBack"
33    title: Optional[str] = None
34    value: Optional[str] = None
35    text: Optional[str] = None
36    display_text: Optional[str] = None
37    image: Optional[str] = None

41class ChannelAccount(_CamelModel):
42    """Represents a user or bot account on a channel.
43
44    Used for the ``from`` and ``recipient`` fields of an activity.
45
46    Attributes:
47        id: Unique account identifier on the channel.
48        name: Display name of the account.
49        aad_object_id: Azure AD object ID (when available).
50        role: Account role (``"bot"`` or ``"user"``).
51    """
52
53    id: str
54    name: Optional[str] = None
55    aad_object_id: Optional[str] = None
56    role: Optional[str] = None

46class ChannelInfo(_CamelModel):
47    """Teams channel information.
48
49    Attributes:
50        id: Unique channel identifier.
51        name: Display name of the channel.
52    """
53
54    id: Optional[str] = None
55    name: Optional[str] = None

71class Conversation(_CamelModel):
72    """Represents a conversation (chat, channel, or group) on a channel.
73
74    Attributes:
75        id: Unique conversation identifier on the channel.
76    """
77
78    id: str

 46class ConversationClient:
 47    """Typed client for Bot Service Conversation REST API operations.
 48
 49    All methods accept a ``service_url`` and ``conversation_id`` to target
 50    the correct channel endpoint.  Authentication is handled automatically
 51    via the injected :class:`TokenProvider`.
 52    """
 53
 54    def __init__(self, get_token: Optional[TokenProvider] = None) -> None:
 55        """Initialise the conversation client.
 56
 57        Args:
 58            get_token: Async callable that supplies a bearer token.
 59                When ``None``, requests are unauthenticated.
 60        """
 61        self._http = _BotHttpClient(get_token)
 62
 63    async def send_activity_async(
 64        self,
 65        service_url: str,
 66        conversation_id: str,
 67        activity: Union[
 68            CoreActivity,
 69            dict[str, Any],
 70        ],
 71    ) -> Optional[ResourceResponse]:
 72        """Send an activity to a conversation.
 73
 74        Args:
 75            service_url: The channel's service URL.
 76            conversation_id: Target conversation identifier.
 77            activity: Activity payload (model or dict).
 78
 79        Returns:
 80            A :class:`ResourceResponse` with the new activity ID, or ``None``.
 81        """
 82        tracer = get_tracer()
 83        metrics = get_metrics()
 84        if metrics:
 85            metrics.outbound_calls.add(1, {"operation": "sendActivity"})
 86        if tracer:
 87            with tracer.start_as_current_span("botas.conversation_client") as span:
 88                span.set_attribute("conversation.id", conversation_id)
 89                span.set_attribute("activity.type", getattr(activity, "type", "") or "")
 90                span.set_attribute("service.url", service_url)
 91                try:
 92                    result = await self._do_send(service_url, conversation_id, activity)
 93                    if result:
 94                        span.set_attribute("activity.id", result.id or "")
 95                    return result
 96                except Exception:
 97                    if metrics:
 98                        metrics.outbound_errors.add(1, {"operation": "sendActivity"})
 99                    raise
100        try:
101            return await self._do_send(service_url, conversation_id, activity)
102        except Exception:
103            if metrics:
104                metrics.outbound_errors.add(1, {"operation": "sendActivity"})
105            raise
106
107    async def _do_send(
108        self,
109        service_url: str,
110        conversation_id: str,
111        activity: Union[
112            CoreActivity,
113            dict[str, Any],
114        ],
115    ) -> Optional[ResourceResponse]:
116        """Execute the actual HTTP POST for sending an activity."""
117        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}/activities"
118        data = await self._http.post(
119            service_url,
120            endpoint,
121            _serialize(activity),
122            _BotRequestOptions(operation_description="send activity"),
123        )
124        return ResourceResponse.model_validate(data) if data else None
125
126    async def update_activity_async(
127        self,
128        service_url: str,
129        conversation_id: str,
130        activity_id: str,
131        activity: Union[
132            CoreActivity,
133            dict[str, Any],
134        ],
135    ) -> Optional[ResourceResponse]:
136        """Update an existing activity in a conversation.
137
138        Args:
139            service_url: The channel's service URL.
140            conversation_id: Conversation containing the activity.
141            activity_id: ID of the activity to update.
142            activity: Replacement activity payload.
143
144        Returns:
145            A :class:`ResourceResponse`, or ``None``.
146        """
147        metrics = get_metrics()
148        if metrics:
149            metrics.outbound_calls.add(1, {"operation": "updateActivity"})
150        try:
151            endpoint = (
152                f"/v3/conversations/{_encode_conversation_id(conversation_id)}/activities/{_encode_id(activity_id)}"
153            )
154            data = await self._http.put(
155                service_url,
156                endpoint,
157                _serialize(activity),
158                _BotRequestOptions(operation_description="update activity"),
159            )
160            return ResourceResponse.model_validate(data) if data else None
161        except Exception:
162            if metrics:
163                metrics.outbound_errors.add(1, {"operation": "updateActivity"})
164            raise
165
166    async def delete_activity_async(self, service_url: str, conversation_id: str, activity_id: str) -> None:
167        """Delete an activity from a conversation.
168
169        Args:
170            service_url: The channel's service URL.
171            conversation_id: Conversation containing the activity.
172            activity_id: ID of the activity to delete.
173        """
174        metrics = get_metrics()
175        if metrics:
176            metrics.outbound_calls.add(1, {"operation": "deleteActivity"})
177        try:
178            endpoint = (
179                f"/v3/conversations/{_encode_conversation_id(conversation_id)}/activities/{_encode_id(activity_id)}"
180            )
181            await self._http.delete(
182                service_url,
183                endpoint,
184                _BotRequestOptions(operation_description="delete activity"),
185            )
186        except Exception:
187            if metrics:
188                metrics.outbound_errors.add(1, {"operation": "deleteActivity"})
189            raise
190
191    async def get_conversation_members_async(self, service_url: str, conversation_id: str) -> list[ChannelAccount]:
192        """Retrieve all members of a conversation.
193
194        Args:
195            service_url: The channel's service URL.
196            conversation_id: Target conversation identifier.
197
198        Returns:
199            List of :class:`ChannelAccount` objects for each member.
200        """
201        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}/members"
202        data = await self._http.get(
203            service_url,
204            endpoint,
205            options=_BotRequestOptions(operation_description="get conversation members"),
206        )
207        return [ChannelAccount.model_validate(m) for m in (data or [])]
208
209    async def get_conversation_member_async(
210        self, service_url: str, conversation_id: str, member_id: str
211    ) -> Optional[ChannelAccount]:
212        """Retrieve a single conversation member by ID.
213
214        Args:
215            service_url: The channel's service URL.
216            conversation_id: Target conversation identifier.
217            member_id: The member's account ID.
218
219        Returns:
220            A :class:`ChannelAccount`, or ``None`` if the member is not found.
221        """
222        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}/members/{_encode_id(member_id)}"
223        data = await self._http.get(
224            service_url,
225            endpoint,
226            options=_BotRequestOptions(operation_description="get conversation member", return_none_on_not_found=True),
227        )
228        return ChannelAccount.model_validate(data) if data else None
229
230    async def get_conversation_paged_members_async(
231        self,
232        service_url: str,
233        conversation_id: str,
234        page_size: Optional[int] = None,
235        continuation_token: Optional[str] = None,
236    ) -> _PagedMembersResult:
237        """Retrieve conversation members with server-side pagination.
238
239        Args:
240            service_url: The channel's service URL.
241            conversation_id: Target conversation identifier.
242            page_size: Maximum members per page (channel may enforce its own limit).
243            continuation_token: Opaque token from a previous page to fetch the next.
244
245        Returns:
246            A :class:`_PagedMembersResult` containing members and an optional
247            continuation token for the next page.
248        """
249        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}/pagedmembers"
250        params = {
251            "pageSize": str(page_size) if page_size else None,
252            "continuationToken": continuation_token,
253        }
254        data = await self._http.get(
255            service_url,
256            endpoint,
257            params=params,
258            options=_BotRequestOptions(operation_description="get paged members"),
259        )
260        return _PagedMembersResult.model_validate(data) if data else _PagedMembersResult()
261
262    async def delete_conversation_member_async(self, service_url: str, conversation_id: str, member_id: str) -> None:
263        """Remove a member from a conversation.
264
265        Args:
266            service_url: The channel's service URL.
267            conversation_id: Target conversation identifier.
268            member_id: The member's account ID.
269        """
270        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}/members/{_encode_id(member_id)}"
271        await self._http.delete(
272            service_url,
273            endpoint,
274            _BotRequestOptions(operation_description="delete conversation member"),
275        )
276
277    async def create_conversation_async(
278        self, service_url: str, parameters: _ConversationParameters
279    ) -> Optional[_ConversationResourceResponse]:
280        """Create a new conversation on the channel.
281
282        Args:
283            service_url: The channel's service URL.
284            parameters: Conversation creation parameters (members, topic, etc.).
285
286        Returns:
287            A :class:`_ConversationResourceResponse` with the new conversation
288            ID and service URL, or ``None``.
289        """
290        data = await self._http.post(
291            service_url,
292            "/v3/conversations",
293            _serialize(parameters),
294            _BotRequestOptions(operation_description="create conversation"),
295        )
296        return _ConversationResourceResponse.model_validate(data) if data else None
297
298    async def get_conversations_async(
299        self, service_url: str, continuation_token: Optional[str] = None
300    ) -> _ConversationsResult:
301        """List conversations the bot has participated in.
302
303        Args:
304            service_url: The channel's service URL.
305            continuation_token: Opaque token from a previous page.
306
307        Returns:
308            A :class:`_ConversationsResult` with conversations and an optional
309            continuation token.
310        """
311        params = {"continuationToken": continuation_token}
312        data = await self._http.get(
313            service_url,
314            "/v3/conversations",
315            params=params,
316            options=_BotRequestOptions(operation_description="get conversations"),
317        )
318        return _ConversationsResult.model_validate(data) if data else _ConversationsResult()
319
320    async def send_conversation_history_async(
321        self, service_url: str, conversation_id: str, _Transcript: _Transcript
322    ) -> Optional[ResourceResponse]:
323        """Upload a _Transcript of activities to a conversation's history.
324
325        Args:
326            service_url: The channel's service URL.
327            conversation_id: Target conversation identifier.
328            _Transcript: A :class:`_Transcript` containing activities to upload.
329
330        Returns:
331            A :class:`ResourceResponse`, or ``None``.
332        """
333        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}/activities/history"
334        data = await self._http.post(
335            service_url,
336            endpoint,
337            _serialize(_Transcript),
338            _BotRequestOptions(operation_description="send conversation history"),
339        )
340        return ResourceResponse.model_validate(data) if data else None
341
342    async def get_conversation_account_async(self, service_url: str, conversation_id: str) -> Optional[Conversation]:
343        """Retrieve the conversation account details.
344
345        Args:
346            service_url: The channel's service URL.
347            conversation_id: Target conversation identifier.
348
349        Returns:
350            A :class:`Conversation` object, or ``None`` if not found.
351        """
352        endpoint = f"/v3/conversations/{_encode_conversation_id(conversation_id)}"
353        data = await self._http.get(
354            service_url,
355            endpoint,
356            options=_BotRequestOptions(operation_description="get conversation", return_none_on_not_found=True),
357        )
358        return Conversation.model_validate(data) if data else None
359
360    async def aclose(self) -> None:
361        """Close the underlying HTTP client and release resources."""
362        await self._http.aclose()