diff --git a/examples/http-adapters/README.md b/examples/http-adapters/README.md new file mode 100644 index 00000000..465af849 --- /dev/null +++ b/examples/http-adapters/README.md @@ -0,0 +1,55 @@ +# HTTP Adapters Examples + +Examples showing how to use custom `HttpServerAdapter` implementations and non-managed server patterns with the Teams Python SDK. + +## Examples + +### 1. Starlette Adapter (`starlette_echo.py`) + +A custom `HttpServerAdapter` implementation for [Starlette](https://www.starlette.io/). Demonstrates how to write an adapter for any ASGI framework. + +**Pattern**: Custom adapter, SDK-managed server lifecycle (`app.start()`) + +```bash +python src/starlette_echo.py +``` + +### 2. Non-Managed FastAPI (`fastapi_non_managed.py`) + +Use your own FastAPI app with your own routes, and let the SDK register `/api/messages` on it. You manage the server lifecycle yourself. + +**Pattern**: Default `FastAPIAdapter` with user-provided FastAPI instance, user-managed server (`app.initialize()` + your own `uvicorn.Server`) + +```bash +python src/fastapi_non_managed.py +``` + +## Key Concepts + +### Managed vs Non-Managed + +| | Managed | Non-Managed | +|---|---|---| +| **Entry point** | `app.start(port)` | `app.initialize()` + start server yourself | +| **Who starts the server** | The SDK (via adapter) | You | +| **When to use** | New apps, simple setup | Existing apps, custom server config | + +### Writing a Custom Adapter + +Implement the `HttpServerAdapter` protocol: + +```python +class MyAdapter: + def register_route(self, method, path, handler): ... + def serve_static(self, path, directory): ... + async def start(self, port): ... + async def stop(self): ... +``` + +The handler signature is framework-agnostic: + +```python +async def handler(request: HttpRequest) -> HttpResponse: + # request = { "body": dict, "headers": dict } + # return { "status": int, "body": object } +``` diff --git a/examples/http-adapters/pyproject.toml b/examples/http-adapters/pyproject.toml new file mode 100644 index 00000000..e8bbacd2 --- /dev/null +++ b/examples/http-adapters/pyproject.toml @@ -0,0 +1,17 @@ +[project] +name = "http-adapters" +version = "0.1.0" +description = "Examples showing custom HttpServerAdapter and non-managed server patterns" +readme = "README.md" +requires-python = ">=3.12,<3.15" +dependencies = [ + "dotenv>=0.9.9", + "microsoft-teams-apps", + "microsoft-teams-api", + "starlette", + "uvicorn", + "httptools", +] + +[tool.uv.sources] +microsoft-teams-apps = { workspace = true } diff --git a/examples/http-adapters/src/fastapi_non_managed.py b/examples/http-adapters/src/fastapi_non_managed.py new file mode 100644 index 00000000..3ac8d206 --- /dev/null +++ b/examples/http-adapters/src/fastapi_non_managed.py @@ -0,0 +1,91 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. + +Non-Managed FastAPI Server +========================== +Teams echo bot where YOU manage the server lifecycle. + +This demonstrates the "non-managed" pattern — you create your own FastAPI app +with your own routes, wrap it in a FastAPIAdapter, call app.initialize() to +register the Teams routes, then start uvicorn yourself. + +This is ideal when: +- You have an existing FastAPI app and want to add Teams bot support +- You need full control over server configuration (TLS, workers, middleware) +- You're deploying to a platform that manages the server (e.g. Azure Functions) + +Run: + python src/fastapi_non_managed.py +""" + +import asyncio +import os + +import uvicorn +from fastapi import FastAPI +from fastapi.responses import HTMLResponse +from microsoft_teams.api import MessageActivity +from microsoft_teams.apps import ActivityContext, App, FastAPIAdapter + +# 1. Create your own FastAPI app with your own routes +my_fastapi = FastAPI(title="My App + Teams Bot") + + +@my_fastapi.get("/health") +async def health(): + return {"status": "healthy"} + + +@my_fastapi.get("/api/users") +async def users(): + return {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]} + + +@my_fastapi.get("/") +async def homepage(): + return HTMLResponse(""" +

FastAPI + Teams Bot

+

Your FastAPI server is running with a Teams bot!

+ + """) + + +# 2. Create a FastAPIAdapter wrapping your existing FastAPI app +adapter = FastAPIAdapter(app=my_fastapi) + +# 3. Create the Teams app with the adapter +app = App(http_server_adapter=adapter) + + +# 4. Handle incoming messages +@app.on_message +async def handle_message(ctx: ActivityContext[MessageActivity]): + await ctx.send(f"[FastAPI non-managed] You said: '{ctx.activity.text}'") + + +async def main(): + port = int(os.getenv("PORT", "3978")) + + # 5. Initialize only — registers /api/messages on our FastAPI app + # Does NOT start a server + await app.initialize() + + print(f"Starting server on http://localhost:{port}") + print(" GET / — Homepage") + print(" GET /health — Health check") + print(" GET /api/users — Users API") + print(" POST /api/messages — Teams bot endpoint (added by SDK)") + + # 6. Start your own uvicorn server + config = uvicorn.Config(app=my_fastapi, host="0.0.0.0", port=port, log_level="info") + server = uvicorn.Server(config) + await server.serve() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/http-adapters/src/starlette_adapter.py b/examples/http-adapters/src/starlette_adapter.py new file mode 100644 index 00000000..0dbab016 --- /dev/null +++ b/examples/http-adapters/src/starlette_adapter.py @@ -0,0 +1,91 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. + +Starlette Adapter +================= +A custom HttpServerAdapter implementation for Starlette. + +This shows how to implement the adapter protocol for any ASGI framework. +The adapter translates between the framework's request/response model +and the SDK's pure handler pattern: ({ body, headers }) -> { status, body }. +""" + +from typing import Optional + +import uvicorn +from microsoft_teams.apps.http.adapter import HttpMethod, HttpRequest, HttpResponse, HttpRouteHandler +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse, Response +from starlette.routing import Mount, Route +from starlette.staticfiles import StaticFiles + + +class StarletteAdapter: + """ + HttpServerAdapter implementation wrapping Starlette + uvicorn. + + Usage: + adapter = StarletteAdapter() + app = App(http_server_adapter=adapter) + await app.start(3978) + + Or bring your own Starlette instance: + starlette_app = Starlette() + adapter = StarletteAdapter(starlette_app) + app = App(http_server_adapter=adapter) + await app.initialize() # Just registers routes, doesn't start server + """ + + def __init__(self, app: Optional[Starlette] = None): + self._app = app or Starlette() + self._is_user_provided = app is not None + self._server: Optional[uvicorn.Server] = None + self._routes: list[Route] = [] + + @property + def app(self) -> Starlette: + """The underlying Starlette instance.""" + return self._app + + def register_route(self, method: HttpMethod, path: str, handler: HttpRouteHandler) -> None: + """Register a route handler on the Starlette app.""" + + async def starlette_handler(request: Request) -> Response: + body = await request.json() + headers = dict(request.headers) + http_request = HttpRequest(body=body, headers=headers) + result: HttpResponse = await handler(http_request) + status = result["status"] + resp_body = result.get("body") + if resp_body is not None: + return JSONResponse(content=resp_body, status_code=status) + return Response(status_code=status) + + route = Route(path, starlette_handler, methods=[method]) + self._routes.append(route) + self._app.routes.insert(0, route) + + def serve_static(self, path: str, directory: str) -> None: + """Mount a static files directory.""" + name = path.strip("/").replace("/", "-") or "static" + mount = Mount(path, app=StaticFiles(directory=directory, check_dir=True, html=True), name=name) + self._app.routes.append(mount) + + async def start(self, port: int) -> None: + """Start the uvicorn server. Blocks until stopped.""" + if self._is_user_provided: + raise RuntimeError( + "Cannot call start() when a Starlette instance was provided by user. " + "Manage the server lifecycle yourself." + ) + + config = uvicorn.Config(app=self._app, host="0.0.0.0", port=port, log_level="info") + self._server = uvicorn.Server(config) + await self._server.serve() + + async def stop(self) -> None: + """Signal the server to stop.""" + if self._server: + self._server.should_exit = True diff --git a/examples/http-adapters/src/starlette_echo.py b/examples/http-adapters/src/starlette_echo.py new file mode 100644 index 00000000..8b0e3960 --- /dev/null +++ b/examples/http-adapters/src/starlette_echo.py @@ -0,0 +1,45 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. + +Starlette Echo Bot +================== +Teams echo bot using a custom StarletteAdapter. + +This demonstrates the "managed" pattern — the SDK manages the server lifecycle +via app.start(). The adapter creates its own Starlette app and uvicorn server. + +Run: + python src/starlette_echo.py +""" + +import asyncio + +from microsoft_teams.api import MessageActivity +from microsoft_teams.apps import ActivityContext, App +from starlette_adapter import StarletteAdapter + +# 1. Create adapter +adapter = StarletteAdapter() + + +# 2. Add custom routes directly on the Starlette instance +@adapter.app.route("/health") +async def health(request): + from starlette.responses import JSONResponse + + return JSONResponse({"status": "healthy"}) + + +# 3. Create the Teams app with the adapter +app = App(http_server_adapter=adapter) + + +# 4. Handle incoming messages +@app.on_message +async def handle_message(ctx: ActivityContext[MessageActivity]): + await ctx.send(f"[Starlette] You said: '{ctx.activity.text}'") + + +if __name__ == "__main__": + asyncio.run(app.start()) diff --git a/examples/proactive-messaging/pyproject.toml b/examples/proactive-messaging/pyproject.toml index 437354e7..800f9712 100644 --- a/examples/proactive-messaging/pyproject.toml +++ b/examples/proactive-messaging/pyproject.toml @@ -3,7 +3,7 @@ name = "proactive-messaging" version = "0.1.0" description = "Example showing proactive messaging without running a server" readme = "README.md" -requires-python = ">=3.12,<3.14" +requires-python = ">=3.12,<3.15" dependencies = [ "dotenv>=0.9.9", "microsoft-teams-apps", diff --git a/packages/a2aprotocol/src/microsoft_teams/a2a/server/plugin.py b/packages/a2aprotocol/src/microsoft_teams/a2a/server/plugin.py index 5c106bd1..47f71de8 100644 --- a/packages/a2aprotocol/src/microsoft_teams/a2a/server/plugin.py +++ b/packages/a2aprotocol/src/microsoft_teams/a2a/server/plugin.py @@ -9,7 +9,8 @@ from microsoft_teams.apps import ( DependencyMetadata, EventMetadata, - HttpPlugin, + FastAPIAdapter, + HttpServer, LoggerDependencyOptions, Plugin, PluginBase, @@ -30,7 +31,7 @@ @Plugin(name="a2a", version="0.3.7", description="A2A Server Plugin") class A2APlugin(PluginBase): logger: Annotated[Logger, LoggerDependencyOptions()] - http: Annotated[HttpPlugin, DependencyMetadata()] + server: Annotated[HttpServer, DependencyMetadata()] emit: Annotated[Callable[[str, A2AMessageEvent], Awaitable[None]], EventMetadata(name="custom")] @@ -76,7 +77,10 @@ async def on_init(self) -> None: self.logger.info(f"A2A agent set up at {self.agent_card_path}") self.logger.info(f"A2A agent listening at {self.path}") - self.http.app.mount(self.path, self.app) + adapter = self.server.adapter + if not isinstance(adapter, FastAPIAdapter): + raise RuntimeError("A2APlugin requires FastAPIAdapter. Custom adapters are not supported.") + adapter.app.mount(self.path, self.app) def _setup_executor(self) -> AgentExecutor: return CustomAgentExecutor(self.emit) diff --git a/packages/apps/src/microsoft_teams/apps/__init__.py b/packages/apps/src/microsoft_teams/apps/__init__.py index 2aa32332..265126f1 100644 --- a/packages/apps/src/microsoft_teams/apps/__init__.py +++ b/packages/apps/src/microsoft_teams/apps/__init__.py @@ -8,14 +8,22 @@ from .auth import * # noqa: F403 from .contexts import * # noqa: F403 from .events import * # noqa: F401, F403 -from .http_plugin import HttpPlugin +from .http import FastAPIAdapter, HttpServer, HttpServerAdapter from .http_stream import HttpStream from .options import AppOptions from .plugins import * # noqa: F401, F403 from .routing import ActivityContext # Combine all exports from submodules -__all__: list[str] = ["App", "AppOptions", "HttpPlugin", "HttpStream", "ActivityContext"] +__all__: list[str] = [ + "App", + "AppOptions", + "HttpServer", + "HttpServerAdapter", + "FastAPIAdapter", + "HttpStream", + "ActivityContext", +] __all__.extend(auth.__all__) __all__.extend(events.__all__) __all__.extend(plugins.__all__) diff --git a/packages/apps/src/microsoft_teams/apps/app.py b/packages/apps/src/microsoft_teams/apps/app.py index f72aceee..6e55f35b 100644 --- a/packages/apps/src/microsoft_teams/apps/app.py +++ b/packages/apps/src/microsoft_teams/apps/app.py @@ -11,7 +11,6 @@ from dependency_injector import providers from dotenv import find_dotenv, load_dotenv -from fastapi import Request from microsoft_teams.api import ( Account, ActivityBase, @@ -35,7 +34,7 @@ from .app_plugins import PluginProcessor from .app_process import ActivityProcessor from .auth import TokenValidator -from .auth.remote_function_jwt_middleware import remote_function_jwt_validation +from .auth.remote_function_jwt_middleware import validate_remote_function_request from .container import Container from .contexts.function_context import FunctionContext from .events import ( @@ -46,9 +45,10 @@ get_event_type_from_signature, is_registered_event, ) -from .http_plugin import HttpPlugin +from .http import FastAPIAdapter, HttpServer +from .http.adapter import HttpRequest, HttpResponse from .options import AppOptions, InternalAppOptions -from .plugins import PluginBase, PluginStartEvent, get_metadata +from .plugins import PluginBase, PluginStartEvent from .routing import ActivityHandlerMixin, ActivityRouter from .routing.activity_context import ActivityContext from .token_manager import TokenManager @@ -111,22 +111,13 @@ def __init__(self, **options: Unpack[AppOptions]): plugins: List[PluginBase] = list(self.options.plugins) - http_plugin = None - for i, plugin in enumerate(plugins): - meta = get_metadata(type(plugin)) - if meta.name == "http": - http_plugin = plugin - plugins.pop(i) - break - - if not http_plugin: - http_plugin = HttpPlugin(logger=self.log, skip_auth=self.options.skip_auth) - - plugins.insert(0, http_plugin) - self.http = cast(HttpPlugin, http_plugin) + # Create HttpServer (not a plugin — owned directly by App) + adapter = self.options.http_server_adapter or FastAPIAdapter() + self.server = HttpServer(adapter, self.log) + self.container.set_provider("HttpServer", providers.Object(self.server)) + self.container.set_provider("server", providers.Object(self.server)) self._port: Optional[int] = None - self._running = False self._initialized = False # initialize ActivitySender for sending activities @@ -173,11 +164,6 @@ def port(self) -> Optional[int]: """Port the app is running on.""" return self._port - @property - def is_running(self) -> bool: - """Whether the app is currently running.""" - return self._running - @property def logger(self) -> Logger: """The logger instance used by the app.""" @@ -212,12 +198,16 @@ async def initialize(self) -> None: return try: + # Initialize plugins first (they may register routes, e.g. BotBuilder's /api/messages) for plugin in self.plugins: - # Inject the dependencies self._plugin_processor.inject(plugin) if hasattr(plugin, "on_init") and callable(plugin.on_init): await plugin.on_init() + # Initialize HttpServer (JWT validation + default /api/messages route) + self.server.on_request = self._process_activity_event + self.server.initialize(credentials=self.credentials, skip_auth=self.options.skip_auth) + self._initialized = True self.log.info("Teams app initialized successfully (without HTTP server)") @@ -226,6 +216,11 @@ async def initialize(self) -> None: self._events.emit("error", ErrorEvent(error, context={"method": "initialize"})) raise + async def _process_activity_event(self, event: Any) -> Any: + """Process an activity event through the app pipeline. Used as HttpServer.on_request callback.""" + await self.event_manager.on_activity(event) + return await self.activity_processor.process_activity(self.plugins, event) + async def start(self, port: Optional[int] = None) -> None: """ Start the Teams application and begin serving HTTP requests. @@ -236,10 +231,6 @@ async def start(self, port: Optional[int] = None) -> None: Args: port: Port to listen on (defaults to PORT env var or 3978) """ - if self._running: - self.log.warning("App is already running") - return - self._port = port or int(os.getenv("PORT", "3978")) try: @@ -247,49 +238,39 @@ async def start(self, port: Optional[int] = None) -> None: if not self._initialized: await self.initialize() - # Set callback and start HTTP plugin - async def on_http_ready() -> None: - self.log.info("Teams app started successfully") - assert self._port is not None, "Port must be set before emitting start event" - self._events.emit("start", StartEvent(port=self._port)) - self._running = True - - self.http.on_ready_callback = on_http_ready - + # Start plugins and HTTP server concurrently (both may block with serve()) tasks: List[Awaitable[Any]] = [] event = PluginStartEvent(port=self._port) for plugin in self.plugins: is_callable = hasattr(plugin, "on_start") and callable(plugin.on_start) if is_callable: tasks.append(plugin.on_start(event)) + + self.log.info("Teams app started successfully") + self._events.emit("start", StartEvent(port=self._port)) + + tasks.append(self.server.start(self._port)) await asyncio.gather(*tasks) except Exception as error: - self._running = False self.log.error(f"Failed to start app: {error}") self._events.emit("error", ErrorEvent(error, context={"method": "start", "port": self._port})) raise async def stop(self) -> None: """Stop the Teams application.""" - if not self._running: - return - try: - # Set callback and stop HTTP plugin first - async def on_http_stopped() -> None: - # Stop all other plugins after HTTP is stopped - for plugin in reversed(self.plugins): - is_callable = hasattr(plugin, "on_stop") and callable(plugin.on_stop) - if plugin is not self.http and is_callable: - await plugin.on_stop() + # Stop HTTP server first + await self.server.stop() - self._running = False - self.log.info("Teams app stopped") - self._events.emit("stop", StopEvent()) + # Stop all plugins + for plugin in reversed(self.plugins): + is_callable = hasattr(plugin, "on_stop") and callable(plugin.on_stop) + if is_callable: + await plugin.on_stop() - self.http.on_stopped_callback = on_http_stopped - await self.http.on_stop() + self.log.info("Teams app stopped") + self._events.emit("stop", StopEvent()) except Exception as error: self.log.error(f"Failed to stop app: {error}") @@ -475,7 +456,7 @@ def page(self, name: str, dir_path: str, page_path: Optional[str] = None) -> Non app.page("customform", os.path.join(os.path.dirname(__file__), "views", "customform"), "/tabs/dialog-form") ``` """ - self.http.mount(name, dir_path, page_path=page_path) + self.server.serve_static(page_path or f"/{name}", dir_path) def tab(self, name: str, path: str) -> None: """ @@ -511,23 +492,25 @@ def decorator(func: FCtx) -> FCtx: endpoint_name = name_or_func if isinstance(name_or_func, str) else func.__name__.replace("_", "-") self.logger.debug("Generated endpoint name for function '%s': %s", func.__name__, endpoint_name) - async def endpoint(req: Request): - middleware = remote_function_jwt_validation(self.log, self.entra_token_validator) - - async def call_next(r: Request) -> Any: - ctx = FunctionContext( - id=self.id, - api=self.api, - activity_sender=self.activity_sender, - log=self.log, - data=await r.json(), - **r.state.context.__dict__, - ) - return await func(ctx) - - return await middleware(req, call_next) + async def handler(request: HttpRequest) -> HttpResponse: + client_context, error = await validate_remote_function_request( + request["headers"], self.log, self.entra_token_validator + ) + if error or not client_context: + return HttpResponse(status=401, body={"detail": error or "unauthorized"}) + + ctx = FunctionContext( + id=self.id, + api=self.api, + activity_sender=self.activity_sender, + log=self.log, + data=request["body"], + **client_context.__dict__, + ) + result = await func(ctx) + return HttpResponse(status=200, body=result) - self.http.post(f"/api/functions/{endpoint_name}")(endpoint) + self.server.register_route("POST", f"/api/functions/{endpoint_name}", handler) return func # Direct decoration: @app.func diff --git a/packages/apps/src/microsoft_teams/apps/auth/__init__.py b/packages/apps/src/microsoft_teams/apps/auth/__init__.py index 6fa5f743..46b58cfe 100644 --- a/packages/apps/src/microsoft_teams/apps/auth/__init__.py +++ b/packages/apps/src/microsoft_teams/apps/auth/__init__.py @@ -4,7 +4,7 @@ """ from .jwt_middleware import create_jwt_validation_middleware -from .remote_function_jwt_middleware import remote_function_jwt_validation +from .remote_function_jwt_middleware import validate_remote_function_request from .token_validator import TokenValidator -__all__ = ["TokenValidator", "create_jwt_validation_middleware", "remote_function_jwt_validation"] +__all__ = ["TokenValidator", "create_jwt_validation_middleware", "validate_remote_function_request"] diff --git a/packages/apps/src/microsoft_teams/apps/auth/remote_function_jwt_middleware.py b/packages/apps/src/microsoft_teams/apps/auth/remote_function_jwt_middleware.py index 5e1b87aa..e9fe1ea8 100644 --- a/packages/apps/src/microsoft_teams/apps/auth/remote_function_jwt_middleware.py +++ b/packages/apps/src/microsoft_teams/apps/auth/remote_function_jwt_middleware.py @@ -4,79 +4,89 @@ """ from logging import Logger -from typing import Any, Awaitable, Callable, Dict, List, Optional - -from fastapi import HTTPException, Request, Response +from typing import Dict, Optional from ..contexts import ClientContext from .token_validator import TokenValidator -def require_fields(fields: Dict[str, Optional[Any]], context: str, logger: Logger) -> None: - missing: List[str] = [name for name, value in fields.items() if not value] +def _require_fields(fields: Dict[str, Optional[str]], context: str, logger: Logger) -> Optional[str]: + """Validate required fields are present. Returns error message if any are missing, None otherwise.""" + missing = [name for name, value in fields.items() if not value] if missing: message = f"Missing or invalid fields in {context}: {', '.join(missing)}" logger.warning(message) - raise HTTPException(status_code=401, detail=message) + return message + return None -def remote_function_jwt_validation(logger: Logger, entra_token_validator: Optional[TokenValidator]): +async def validate_remote_function_request( + headers: Dict[str, str], + logger: Logger, + entra_token_validator: Optional[TokenValidator], +) -> tuple[Optional[ClientContext], Optional[str]]: """ - Middleware to validate JWT for remote function calls. + Validate JWT and extract client context from request headers for remote function calls. + Args: - entra_token_validator: TokenValidator instance for Entra ID tokens - logger: Logger instance + headers: Request headers dict. + logger: Logger instance. + entra_token_validator: TokenValidator instance for Entra ID tokens. Returns: - Middleware function that can be added to FastAPI app + Tuple of (ClientContext, None) on success or (None, error_message) on failure. """ + # Extract auth token + authorization = headers.get("Authorization") or headers.get("authorization") or "" + parts = authorization.split(" ") + auth_token = parts[1] if len(parts) == 2 and parts[0].lower() == "bearer" else "" - async def middleware(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response: - # Extract auth token - authorization = request.headers.get("Authorization", "") - parts = authorization.split(" ") - auth_token = parts[1] if len(parts) == 2 and parts[0].lower() == "bearer" else "" + # Validate headers + error = _require_fields( + { + "X-Teams-App-Session-Id": headers.get("X-Teams-App-Session-Id") or headers.get("x-teams-app-session-id"), + "X-Teams-Page-Id": headers.get("X-Teams-Page-Id") or headers.get("x-teams-page-id"), + "Authorization (Bearer token)": auth_token, + }, + "header", + logger, + ) + if error: + return None, error - # Validate headers - require_fields( - { - "X-Teams-App-Session-Id": request.headers.get("X-Teams-App-Session-Id"), - "X-Teams-Page-Id": request.headers.get("X-Teams-Page-Id"), - "Authorization (Bearer token)": auth_token, - }, - "header", - logger, - ) + if not entra_token_validator: + return None, "Token validator not configured" - if not entra_token_validator: - raise HTTPException(status_code=500, detail="Token validator not configured") + # Validate token + token_payload = await entra_token_validator.validate_token(auth_token) - # Validate token - token_payload = await entra_token_validator.validate_token(auth_token) + # Validate required fields in token + error = _require_fields( + {"oid": token_payload.get("oid"), "tid": token_payload.get("tid"), "name": token_payload.get("name")}, + "token payload", + logger, + ) + if error: + return None, error - # Validate required fields in token - require_fields( - {"oid": token_payload.get("oid"), "tid": token_payload.get("tid"), "name": token_payload.get("name")}, - "token payload", - logger, - ) + def _h(name: str) -> str: + """Get header value case-insensitively.""" + return headers.get(name) or headers.get(name.lower()) or "" - # Build context - request.state.context = ClientContext( - app_session_id=request.headers.get("X-Teams-App-Session-Id"), # type: ignore - tenant_id=token_payload["tid"], - user_id=token_payload["oid"], - user_name=token_payload["name"], - page_id=request.headers.get("X-Teams-Page-Id"), # type: ignore - auth_token=auth_token, # type: ignore - app_id=token_payload.get("appId"), - channel_id=request.headers.get("X-Teams-Channel-Id"), - chat_id=request.headers.get("X-Teams-Chat-Id"), - meeting_id=request.headers.get("X-Teams-Meeting-Id"), - message_id=request.headers.get("X-Teams-Message-Id"), - sub_page_id=request.headers.get("X-Teams-Sub-Page-Id"), - team_id=request.headers.get("X-Teams-Team-Id"), - ) - return await call_next(request) + context = ClientContext( + app_session_id=_h("X-Teams-App-Session-Id"), + tenant_id=token_payload["tid"], + user_id=token_payload["oid"], + user_name=token_payload["name"], + page_id=_h("X-Teams-Page-Id"), + auth_token=auth_token, + app_id=token_payload.get("appId"), + channel_id=_h("X-Teams-Channel-Id") or None, + chat_id=_h("X-Teams-Chat-Id") or None, + meeting_id=_h("X-Teams-Meeting-Id") or None, + message_id=_h("X-Teams-Message-Id") or None, + sub_page_id=_h("X-Teams-Sub-Page-Id") or None, + team_id=_h("X-Teams-Team-Id") or None, + ) - return middleware + return context, None diff --git a/packages/apps/src/microsoft_teams/apps/http/__init__.py b/packages/apps/src/microsoft_teams/apps/http/__init__.py new file mode 100644 index 00000000..f7751fd1 --- /dev/null +++ b/packages/apps/src/microsoft_teams/apps/http/__init__.py @@ -0,0 +1,18 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" + +from .adapter import HttpMethod, HttpRequest, HttpResponse, HttpRouteHandler, HttpServerAdapter +from .fastapi_adapter import FastAPIAdapter +from .http_server import HttpServer + +__all__ = [ + "HttpMethod", + "HttpRequest", + "HttpResponse", + "HttpRouteHandler", + "HttpServer", + "HttpServerAdapter", + "FastAPIAdapter", +] diff --git a/packages/apps/src/microsoft_teams/apps/http/adapter.py b/packages/apps/src/microsoft_teams/apps/http/adapter.py new file mode 100644 index 00000000..bb99c9cb --- /dev/null +++ b/packages/apps/src/microsoft_teams/apps/http/adapter.py @@ -0,0 +1,39 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" + +from typing import Dict, Literal, Protocol, TypedDict, runtime_checkable + +HttpMethod = Literal["POST"] + + +class HttpRequest(TypedDict): + body: Dict[str, object] + headers: Dict[str, str] + + +class HttpResponse(TypedDict): + status: int + body: object + + +class HttpRouteHandler(Protocol): + async def __call__(self, request: HttpRequest) -> HttpResponse: ... + + +@runtime_checkable +class HttpServerAdapter(Protocol): + def register_route(self, method: HttpMethod, path: str, handler: HttpRouteHandler) -> None: + """Register a route handler. Required.""" + ... + + def serve_static(self, path: str, directory: str) -> None: + """Serve static files from a directory. Optional — no-op by default.""" + + async def start(self, port: int) -> None: + """Start the server. Optional — raises if not implemented.""" + raise NotImplementedError("This adapter does not support managed server lifecycle. Start the server yourself.") + + async def stop(self) -> None: + """Stop the server. Optional — no-op by default.""" diff --git a/packages/apps/src/microsoft_teams/apps/http/fastapi_adapter.py b/packages/apps/src/microsoft_teams/apps/http/fastapi_adapter.py new file mode 100644 index 00000000..93da930e --- /dev/null +++ b/packages/apps/src/microsoft_teams/apps/http/fastapi_adapter.py @@ -0,0 +1,76 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" + +from typing import Any, Callable, Dict, Optional + +import uvicorn +from fastapi import FastAPI, Request, Response +from fastapi.responses import JSONResponse +from fastapi.staticfiles import StaticFiles + +from .adapter import HttpMethod, HttpRequest, HttpResponse, HttpRouteHandler + + +class FastAPIAdapter: + """Default HttpServerAdapter implementation wrapping FastAPI + uvicorn.""" + + def __init__( + self, + app: Optional[FastAPI] = None, + server_factory: Optional[Callable[[FastAPI], uvicorn.Server]] = None, + ): + self._fastapi = app or FastAPI() + self._server: Optional[uvicorn.Server] = None + self._server_factory = server_factory + + if server_factory: + self._server = server_factory(self._fastapi) + if self._server.config.app is not self._fastapi: + raise ValueError( + "server_factory must return a uvicorn.Server configured with the provided FastAPI app instance." + ) + + @property + def app(self) -> FastAPI: + """The underlying FastAPI instance.""" + return self._fastapi + + def register_route(self, method: HttpMethod, path: str, handler: HttpRouteHandler) -> None: + """Register a route handler on the FastAPI app.""" + + async def fastapi_handler(request: Request) -> Response: + body: Dict[str, Any] = await request.json() + headers: Dict[str, str] = dict(request.headers) + http_request = HttpRequest(body=body, headers=headers) + result: HttpResponse = await handler(http_request) + status = result["status"] + resp_body = result.get("body") + if resp_body is not None: + return JSONResponse(content=resp_body, status_code=status) + return Response(status_code=status) + + assert method == "POST", f"Unsupported HTTP method: {method}" + self._fastapi.post(path)(fastapi_handler) + + def serve_static(self, path: str, directory: str) -> None: + """Mount a static files directory.""" + name = path.strip("/").replace("/", "-") or "static" + self._fastapi.mount(path, StaticFiles(directory=directory, check_dir=True, html=True), name=name) + + async def start(self, port: int) -> None: + """Start the uvicorn server. Blocks until stopped.""" + if self._server: + if self._server.config.port != port: + pass # User's factory takes precedence + else: + config = uvicorn.Config(app=self._fastapi, host="0.0.0.0", port=port, log_level="info") + self._server = uvicorn.Server(config) + + await self._server.serve() + + async def stop(self) -> None: + """Signal the server to stop.""" + if self._server: + self._server.should_exit = True diff --git a/packages/apps/src/microsoft_teams/apps/http/http_server.py b/packages/apps/src/microsoft_teams/apps/http/http_server.py new file mode 100644 index 00000000..d6b6f410 --- /dev/null +++ b/packages/apps/src/microsoft_teams/apps/http/http_server.py @@ -0,0 +1,178 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" + +from logging import Logger +from types import SimpleNamespace +from typing import Any, Awaitable, Callable, Dict, Optional, cast + +from microsoft_teams.api import Credentials, InvokeResponse, TokenProtocol +from microsoft_teams.api.auth.json_web_token import JsonWebToken +from pydantic import BaseModel + +from ..auth import TokenValidator +from ..events import ActivityEvent, CoreActivity +from .adapter import HttpMethod, HttpRequest, HttpResponse, HttpRouteHandler, HttpServerAdapter + + +class HttpServer: + """ + Core Teams HTTP server. Not a plugin — owned directly by the App. + + Manages an HttpServerAdapter instance and handles JWT validation + and activity processing for the Teams protocol. + """ + + def __init__(self, adapter: HttpServerAdapter, logger: Logger): + self._adapter = adapter + self._logger = logger + self._on_request: Optional[Callable[[ActivityEvent], Awaitable[InvokeResponse[Any]]]] = None + self._token_validator: Optional[TokenValidator] = None + self._skip_auth: bool = False + self._initialized: bool = False + + @property + def adapter(self) -> HttpServerAdapter: + """The underlying HttpServerAdapter.""" + return self._adapter + + @property + def on_request(self) -> Optional[Callable[[ActivityEvent], Awaitable[InvokeResponse[Any]]]]: + """Callback set by App to process activities.""" + return self._on_request + + @on_request.setter + def on_request(self, callback: Optional[Callable[[ActivityEvent], Awaitable[InvokeResponse[Any]]]]) -> None: + self._on_request = callback + + def initialize( + self, + credentials: Optional[Credentials] = None, + skip_auth: bool = False, + ) -> None: + """ + Set up JWT validation and register the default POST /api/messages route. + + Args: + credentials: App credentials for JWT validation. + skip_auth: Whether to skip JWT validation. + """ + if self._initialized: + return + + self._skip_auth = skip_auth + + app_id = getattr(credentials, "client_id", None) if credentials else None + if app_id and not skip_auth: + self._token_validator = TokenValidator.for_service(app_id, self._logger) + self._logger.debug("JWT validation enabled for /api/messages") + + self._adapter.register_route("POST", "/api/messages", self._handle_activity) + self._initialized = True + + async def _handle_activity(self, request: HttpRequest) -> HttpResponse: + """Handle incoming activity on POST /api/messages.""" + try: + body = request["body"] + headers = request["headers"] + + # Validate JWT token + authorization = headers.get("authorization") or headers.get("Authorization") or "" + + if self._token_validator and not self._skip_auth: + if not authorization.startswith("Bearer "): + return HttpResponse(status=401, body={"error": "Unauthorized"}) + + raw_token = authorization.removeprefix("Bearer ") + service_url = cast(Optional[str], body.get("serviceUrl")) + + try: + await self._token_validator.validate_token(raw_token, service_url) + except Exception as e: + self._logger.warning(f"JWT token validation failed: {e}") + return HttpResponse(status=401, body={"error": "Unauthorized"}) + + token: TokenProtocol = cast(TokenProtocol, JsonWebToken(value=raw_token)) + else: + # No auth — use a default token + service_url = cast(Optional[str], body.get("serviceUrl")) + token = cast( + TokenProtocol, + SimpleNamespace( + app_id="", + app_display_name="", + tenant_id="", + service_url=service_url or "", + from_="azure", + from_id="", + is_expired=lambda: False, + ), + ) + + core_activity = CoreActivity.model_validate(body) + activity_type = core_activity.type or "unknown" + activity_id = core_activity.id or "unknown" + self._logger.debug(f"Received activity: {activity_type} (ID: {activity_id})") + + # Process the activity via the App callback + result = await self._process_activity(core_activity, token) + return self._format_response(result) + except Exception as e: + self._logger.exception(str(e)) + return HttpResponse(status=500, body={"error": "Internal server error"}) + + async def _process_activity(self, core_activity: CoreActivity, token: TokenProtocol) -> InvokeResponse[Any]: + """Process an activity via the registered on_request callback.""" + result: InvokeResponse[Any] + try: + event = ActivityEvent(body=core_activity, token=token) + if self._on_request: + result = await self._on_request(event) + else: + self._logger.warning("No on_request handler registered") + result = InvokeResponse(status=500) + except Exception as error: + self._logger.exception(str(error)) + result = InvokeResponse(status=500) + + return result + + def _format_response(self, result: Any) -> HttpResponse: + """Format an InvokeResponse into an HttpResponse.""" + status_code: int = 200 + body: Optional[Any] = None + + resp_dict: Optional[Dict[str, Any]] = None + if isinstance(result, dict): + resp_dict = result + elif isinstance(result, BaseModel): + resp_dict = result.model_dump(exclude_none=True) + + if resp_dict and "status" in resp_dict: + status_code = resp_dict.get("status", 200) + + if resp_dict and "body" in resp_dict: + body = resp_dict.get("body") + + if body is not None: + return HttpResponse(status=status_code, body=body) + return HttpResponse(status=status_code, body=None) + + def register_route(self, method: HttpMethod, path: str, handler: HttpRouteHandler) -> None: + """Delegate route registration to the adapter.""" + self._adapter.register_route(method, path, handler) + + def serve_static(self, path: str, directory: str) -> None: + """Delegate static file serving to the adapter.""" + self._adapter.serve_static(path, directory) + + async def start(self, port: int) -> None: + """Start the HTTP server. Blocks until stopped.""" + self._logger.info(f"Starting HTTP server on port {port}") + await self._adapter.start(port) + + async def stop(self) -> None: + """Stop the HTTP server.""" + self._logger.info("Stopping HTTP server") + await self._adapter.stop() diff --git a/packages/apps/src/microsoft_teams/apps/options.py b/packages/apps/src/microsoft_teams/apps/options.py index 6bd5846c..f56f0dd7 100644 --- a/packages/apps/src/microsoft_teams/apps/options.py +++ b/packages/apps/src/microsoft_teams/apps/options.py @@ -3,6 +3,8 @@ Licensed under the MIT License. """ +from __future__ import annotations + from dataclasses import dataclass, field from logging import Logger from typing import Any, Awaitable, Callable, List, Optional, TypedDict, Union, cast @@ -11,6 +13,7 @@ from microsoft_teams.common import Storage from typing_extensions import Unpack +from .http.adapter import HttpServerAdapter from .plugins import PluginBase @@ -42,6 +45,10 @@ class AppOptions(TypedDict, total=False): plugins: Optional[List[PluginBase]] skip_auth: Optional[bool] + # HTTP adapter + http_server_adapter: Optional[HttpServerAdapter] + """Custom HTTP server adapter. Defaults to FastAPIAdapter if not provided.""" + # OAuth default_connection_name: Optional[str] """The OAuth connection name to use for authentication. Defaults to 'graph'.""" @@ -95,6 +102,8 @@ class InternalAppOptions: Uses environment variable SERVICE_URL if not provided and defaults to https://smba.trafficmanager.net/teams """ + http_server_adapter: Optional[HttpServerAdapter] = None + """Custom HTTP server adapter. Defaults to FastAPIAdapter if not provided.""" @classmethod def from_typeddict(cls, options: AppOptions) -> "InternalAppOptions": diff --git a/packages/apps/tests/test_app.py b/packages/apps/tests/test_app.py index ac41420b..88638b44 100644 --- a/packages/apps/tests/test_app.py +++ b/packages/apps/tests/test_app.py @@ -94,20 +94,17 @@ def basic_options(self, mock_logger, mock_storage): client_secret="test-secret", ) - def _mock_http_plugin(self, app: App) -> App: - """Helper to mock the HTTP plugin's create_stream and close methods.""" - mock_stream = MagicMock() - mock_stream.events = MagicMock() - mock_stream.events.on = MagicMock() - mock_stream.close = AsyncMock() - app.http.create_stream = MagicMock(return_value=mock_stream) + def _mock_http_server(self, app: App) -> App: + """Helper to mock the HTTP server methods.""" + app.server.start = AsyncMock() # type: ignore[method-assign] + app.server.stop = AsyncMock() # type: ignore[method-assign] return app @pytest.fixture(scope="function") def app_with_options(self, basic_options): """Create App with basic options.""" app = App(**basic_options) - return self._mock_http_plugin(app) + return self._mock_http_server(app) @pytest.fixture(scope="function") def app_with_activity_handler(self, mock_logger, mock_storage, mock_activity_handler): @@ -121,51 +118,33 @@ def app_with_activity_handler(self, mock_logger, mock_storage, mock_activity_han ) app = App(**options) app.on_activity(mock_activity_handler) - return self._mock_http_plugin(app) + return self._mock_http_server(app) def test_app_starts_successfully(self, basic_options): """Test that app can be created and initialized.""" app = App(**basic_options) - # Basic functional test - app should be created and not running - assert not app.is_running + # Basic functional test - app should be created assert app.port is None @pytest.mark.asyncio async def test_app_lifecycle_start_stop(self, app_with_options): """Test basic app lifecycle: start and stop.""" - # Mock the underlying HTTP server to avoid actual server startup - async def mock_on_start(event): - # Simulate the HTTP plugin calling the ready callback - if app_with_options.http.on_ready_callback: - await app_with_options.http.on_ready_callback() - - with patch.object(app_with_options.http, "on_start", new_callable=AsyncMock, side_effect=mock_on_start): - # Test start - start_task = asyncio.create_task(app_with_options.start(3978)) - await asyncio.sleep(0.1) - - # App should be running and have correct port - assert app_with_options.is_running - assert app_with_options.port == 3978 - - start_task.cancel() - try: - await start_task - except asyncio.CancelledError: - pass + # Test start — server.start is already mocked by _mock_http_server + start_task = asyncio.create_task(app_with_options.start(3978)) + await asyncio.sleep(0.1) - # Test stop - app_with_options._running = True + assert app_with_options.port == 3978 - async def mock_on_stop(): - if app_with_options.http.on_stopped_callback: - await app_with_options.http.on_stopped_callback() + start_task.cancel() + try: + await start_task + except asyncio.CancelledError: + pass - with patch.object(app_with_options.http, "on_stop", new_callable=AsyncMock, side_effect=mock_on_stop): - await app_with_options.stop() - assert not app_with_options.is_running + # Test stop + await app_with_options.stop() # Event Testing - Focus on functional behavior @@ -187,9 +166,7 @@ async def handle_activity(event: ActivityEvent) -> None: id="test-activity-id", ) - await app_with_activity_handler.event_manager.on_activity( - ActivityEvent(body=core_activity, token=FakeToken()) - ) + await app_with_activity_handler.event_manager.on_activity(ActivityEvent(body=core_activity, token=FakeToken())) # Wait for the async event handler to complete await asyncio.wait_for(event_received.wait(), timeout=1.0) @@ -232,9 +209,7 @@ async def handle_activity_2(event: ActivityEvent) -> None: id="test-activity-id", ) - await app_with_options.event_manager.on_activity( - ActivityEvent(body=core_activity, token=FakeToken()) - ) + await app_with_options.event_manager.on_activity(ActivityEvent(body=core_activity, token=FakeToken())) # Wait for both async event handlers to complete await asyncio.wait_for(both_received.wait(), timeout=1.0) @@ -500,9 +475,9 @@ async def logging_middleware(ctx: ActivityContext) -> None: @pytest.mark.asyncio async def test_func_decorator_registration(self, app_with_options: App): """Simple test that @app.func registers a function.""" - app_with_options.http.post = MagicMock() + mock_register = MagicMock() + app_with_options.server.register_route = mock_register # type: ignore[method-assign] - # Dummy request to simulate a call async def dummy_func(ctx): return "called" @@ -510,8 +485,10 @@ async def dummy_func(ctx): assert decorated == dummy_func # Extract the endpoint path it was registered to - endpoint_path = app_with_options.http.post.call_args[0][0] - assert endpoint_path == f"/api/functions/{dummy_func.__name__.replace('_', '-')}" + mock_register.assert_called_once() + method, path, handler = mock_register.call_args[0] + assert method == "POST" + assert path == f"/api/functions/{dummy_func.__name__.replace('_', '-')}" def test_user_agent_format(self, app_with_options: App): """Test that USER_AGENT follows the expected format teams.py[apps]/{version}.""" diff --git a/packages/apps/tests/test_app_events.py b/packages/apps/tests/test_app_events.py index a4caad5d..a93e9615 100644 --- a/packages/apps/tests/test_app_events.py +++ b/packages/apps/tests/test_app_events.py @@ -13,7 +13,7 @@ ActivityResponseEvent, ActivitySentEvent, ErrorEvent, - HttpPlugin, + PluginBase, ) from microsoft_teams.apps.app_events import EventManager from microsoft_teams.apps.events import CoreActivity @@ -35,12 +35,12 @@ def event_manager(self, mock_event_emitter): @pytest.fixture def mock_plugins(self): - plugin = MagicMock(spec=HttpPlugin) + plugin = MagicMock(spec=PluginBase) plugin.on_error_event = AsyncMock() plugin.on_error = AsyncMock() plugin.on_activity_sent = AsyncMock() plugin.on_activity_response = AsyncMock() - plugin_two = MagicMock(spec=HttpPlugin) + plugin_two = MagicMock(spec=PluginBase) return [plugin, plugin_two] @pytest.mark.asyncio diff --git a/packages/apps/tests/test_app_process.py b/packages/apps/tests/test_app_process.py index 27a66fa0..e760591a 100644 --- a/packages/apps/tests/test_app_process.py +++ b/packages/apps/tests/test_app_process.py @@ -136,7 +136,7 @@ async def test_process_activity_middleware_results(self, activity_processor, mid "conversation": {"id": "conv-789"}, "recipient": {"id": "bot-456", "name": "Test Bot"}, "channelId": "msteams", - } + }, ) mock_token = MagicMock(spec=TokenProtocol) mock_token.service_url = "https://service.url" @@ -172,7 +172,7 @@ async def test_process_activity_raises_exception(self, activity_processor): "conversation": {"id": "conv-789"}, "recipient": {"id": "bot-456", "name": "Test Bot"}, "channelId": "msteams", - } + }, ) mock_token = MagicMock(spec=TokenProtocol) mock_token.service_url = "https://service.url" diff --git a/packages/apps/tests/test_http_server.py b/packages/apps/tests/test_http_server.py new file mode 100644 index 00000000..4ecc3830 --- /dev/null +++ b/packages/apps/tests/test_http_server.py @@ -0,0 +1,241 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" +# pyright: basic + +from typing import cast +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from microsoft_teams.api import ( + ConfigResponse, + InvokeResponse, +) +from microsoft_teams.apps.http import FastAPIAdapter, HttpServer +from microsoft_teams.apps.http.adapter import HttpRequest, HttpResponse + + +class TestHttpServer: + """Test cases for HttpServer public interface.""" + + @pytest.fixture + def mock_logger(self): + """Create a mock logger.""" + return MagicMock() + + @pytest.fixture + def mock_adapter(self): + """Create a mock adapter.""" + adapter = MagicMock() + adapter.register_route = MagicMock() + adapter.serve_static = MagicMock() + adapter.start = AsyncMock() + adapter.stop = AsyncMock() + return adapter + + @pytest.fixture + def server(self, mock_adapter, mock_logger): + """Create HttpServer with mock adapter.""" + return HttpServer(mock_adapter, mock_logger) + + def test_init(self, server, mock_adapter): + """Test HttpServer initialization.""" + assert server.adapter is mock_adapter + assert server.on_request is None + + def test_initialize_registers_route(self, server, mock_adapter): + """Test that initialize registers the /api/messages route.""" + server.initialize() + mock_adapter.register_route.assert_called_once() + call_args = mock_adapter.register_route.call_args + assert call_args[0][0] == "POST" + assert call_args[0][1] == "/api/messages" + + def test_on_request_setter(self, server): + """Test on_request callback setter.""" + + async def handler(event): + return InvokeResponse(status=200) + + server.on_request = handler + assert server.on_request is handler + + @pytest.mark.asyncio + async def test_handle_activity_success(self, server): + """Test successful activity handling.""" + expected_body = {"status": "success"} + expected_response = InvokeResponse(body=cast(ConfigResponse, expected_body), status=200) + + async def mock_handler(event): + return expected_response + + server.on_request = mock_handler + server.initialize(skip_auth=True) + + request = HttpRequest( + body={ + "type": "message", + "id": "test-123", + "text": "Test message", + }, + headers={}, + ) + + result = await server._handle_activity(request) + + assert result["status"] == 200 + assert result["body"] == expected_body + + @pytest.mark.asyncio + async def test_handle_activity_exception(self, server, mock_logger): + """Test activity handling when handler raises exception.""" + + async def failing_handler(event): + raise ValueError("Handler failed") + + server.on_request = failing_handler + server.initialize(skip_auth=True) + + request = HttpRequest( + body={"type": "message", "id": "test-123"}, + headers={}, + ) + + result = await server._handle_activity(request) + + assert result["status"] == 500 + mock_logger.exception.assert_called_once() + + @pytest.mark.asyncio + async def test_handle_activity_no_handler(self, server, mock_logger): + """Test activity handling when no on_request handler is set.""" + server.initialize(skip_auth=True) + + request = HttpRequest( + body={"type": "message", "id": "test-123"}, + headers={}, + ) + + result = await server._handle_activity(request) + + assert result["status"] == 500 + mock_logger.warning.assert_called() + + @pytest.mark.asyncio + async def test_start(self, server, mock_adapter): + """Test server start delegates to adapter.""" + await server.start(3978) + mock_adapter.start.assert_called_once_with(3978) + + @pytest.mark.asyncio + async def test_stop(self, server, mock_adapter): + """Test server stop delegates to adapter.""" + await server.stop() + mock_adapter.stop.assert_called_once() + + def test_register_route_delegates(self, server, mock_adapter): + """Test register_route delegates to adapter.""" + + async def handler(req): + return HttpResponse(status=200, body=None) + + server.register_route("POST", "/custom", handler) + mock_adapter.register_route.assert_called_once_with("POST", "/custom", handler) + + def test_serve_static_delegates(self, server, mock_adapter): + """Test serve_static delegates to adapter.""" + server.serve_static("/static", "/path/to/dir") + mock_adapter.serve_static.assert_called_once_with("/static", "/path/to/dir") + + +class TestFastAPIAdapter: + """Test cases for FastAPIAdapter.""" + + def test_init_creates_fastapi_app(self): + """Test FastAPIAdapter creates a FastAPI app.""" + from fastapi import FastAPI + + adapter = FastAPIAdapter() + assert isinstance(adapter.app, FastAPI) + + def test_register_route(self): + """Test route registration on FastAPI app.""" + adapter = FastAPIAdapter() + + async def handler(req: HttpRequest) -> HttpResponse: + return HttpResponse(status=200, body=None) + + adapter.register_route("POST", "/test", handler) + + # Verify a route was registered on the FastAPI app + routes = [r for r in adapter.app.routes if hasattr(r, "path") and r.path == "/test"] + assert len(routes) == 1 + + def test_serve_static(self, tmp_path): + """Test static file mounting.""" + adapter = FastAPIAdapter() + adapter.serve_static("/static", str(tmp_path)) + + # Verify a mount was registered + mounts = [r for r in adapter.app.routes if hasattr(r, "path") and r.path == "/static"] + assert len(mounts) == 1 + + @pytest.mark.asyncio + async def test_start_creates_uvicorn_server(self): + """Test that start creates and starts a uvicorn server.""" + adapter = FastAPIAdapter() + + mock_server = MagicMock() + mock_server.serve = AsyncMock() + + with ( + patch("microsoft_teams.apps.http.fastapi_adapter.uvicorn.Config") as mock_config, + patch("microsoft_teams.apps.http.fastapi_adapter.uvicorn.Server", return_value=mock_server), + ): + mock_config.return_value = MagicMock() + await adapter.start(3978) + + mock_config.assert_called_once() + mock_server.serve.assert_called_once() + + @pytest.mark.asyncio + async def test_stop(self): + """Test stop signals the server.""" + adapter = FastAPIAdapter() + mock_server = MagicMock() + adapter._server = mock_server + + await adapter.stop() + + assert mock_server.should_exit is True + + @pytest.mark.asyncio + async def test_stop_no_server(self): + """Test stop when no server is running.""" + adapter = FastAPIAdapter() + # Should not raise + await adapter.stop() + + def test_server_factory(self): + """Test custom server factory.""" + mock_server = MagicMock() + + def factory(app): + mock_server.config = MagicMock() + mock_server.config.app = app + return mock_server + + adapter = FastAPIAdapter(server_factory=factory) + assert adapter._server is mock_server + + def test_server_factory_wrong_app_raises(self): + """Test that server factory with wrong app raises.""" + mock_server = MagicMock() + mock_server.config.app = MagicMock() # Different app instance + + def factory(app): + return mock_server + + with pytest.raises(ValueError, match="server_factory must return"): + FastAPIAdapter(server_factory=factory) diff --git a/packages/botbuilder/src/microsoft_teams/botbuilder/botbuilder_plugin.py b/packages/botbuilder/src/microsoft_teams/botbuilder/botbuilder_plugin.py index bc0aae80..5a9baf7c 100644 --- a/packages/botbuilder/src/microsoft_teams/botbuilder/botbuilder_plugin.py +++ b/packages/botbuilder/src/microsoft_teams/botbuilder/botbuilder_plugin.py @@ -6,18 +6,19 @@ import importlib.metadata from logging import Logger from types import SimpleNamespace -from typing import Annotated, Any, Optional, Unpack, cast +from typing import Annotated, Any, Callable, Optional, TypedDict, Unpack, cast -from fastapi import HTTPException, Request, Response -from microsoft_teams.api import Credentials +from microsoft_teams.api import Credentials, InvokeResponse from microsoft_teams.apps import ( DependencyMetadata, - HttpPlugin, + EventMetadata, + HttpServer, LoggerDependencyOptions, Plugin, + PluginBase, ) -from microsoft_teams.apps.events import CoreActivity -from microsoft_teams.apps.http_plugin import HttpPluginOptions +from microsoft_teams.apps.events import ActivityEvent, CoreActivity, ErrorEvent +from microsoft_teams.apps.http import HttpRequest, HttpResponse from botbuilder.core import ( ActivityHandler, @@ -36,15 +37,15 @@ MULTI_TENANT = "multitenant" -class BotBuilderPluginOptions(HttpPluginOptions, total=False): +class BotBuilderPluginOptions(TypedDict, total=False): """Options for configuring the BotBuilder plugin.""" handler: ActivityHandler adapter: CloudAdapter -@Plugin(name="http", version=version, description="BotBuilder plugin for Microsoft Bot Framework integration") -class BotBuilderPlugin(HttpPlugin): +@Plugin(name="botbuilder", version=version, description="BotBuilder plugin for Microsoft Bot Framework integration") +class BotBuilderPlugin(PluginBase): """ BotBuilder plugin that provides Microsoft Bot Framework integration. """ @@ -52,6 +53,10 @@ class BotBuilderPlugin(HttpPlugin): # Dependency injections logger: Annotated[Logger, LoggerDependencyOptions()] credentials: Annotated[Optional[Credentials], DependencyMetadata(optional=True)] + server: Annotated[HttpServer, DependencyMetadata()] + + on_error_event: Annotated[Callable[[ErrorEvent], None], EventMetadata(name="error")] + on_activity_event: Annotated[Callable[[ActivityEvent], InvokeResponse[Any]], EventMetadata(name="activity")] def __init__(self, **options: Unpack[BotBuilderPluginOptions]): """ @@ -60,16 +65,12 @@ def __init__(self, **options: Unpack[BotBuilderPluginOptions]): Args: options: Configuration options for the plugin """ - + super().__init__() self.handler: Optional[ActivityHandler] = options.get("handler") self.adapter: Optional[CloudAdapter] = options.get("adapter") - super().__init__(**options) - async def on_init(self) -> None: """Initialize the plugin when the app starts.""" - await super().on_init() - if not self.adapter: # Extract credentials for Bot Framework authentication client_id: Optional[str] = None @@ -93,46 +94,73 @@ async def on_init(self) -> None: self.logger.debug("BotBuilder plugin initialized successfully") - async def on_activity_request(self, core_activity: CoreActivity, request: Request, response: Response) -> Any: - """ - Handles an incoming activity. + # Register the activity route via HttpServer + self.server.register_route("POST", "/api/messages", self._handle_activity) - Overrides the base HTTP plugin behavior to: - 1. Process the activity using the Bot Framework adapter/handler. - 2. Then pass the request to the parent Teams plugin pipeline. + async def _handle_activity(self, request: HttpRequest) -> HttpResponse: + """ + Pure handler for POST /api/messages. - Returns the final HTTP response. + Processes via Bot Framework, then passes to the Teams pipeline. """ if not self.adapter: raise RuntimeError("plugin not registered") + body = request["body"] + headers = request["headers"] + try: - # Parse activity data from core_activity - activity_dict = core_activity.model_dump(by_alias=True, exclude_none=True) - activity_bf = cast(Activity, Activity().deserialize(activity_dict)) + # Parse activity from body + activity_bf = cast(Activity, Activity().deserialize(body)) - # A POST request must contain an Activity if not activity_bf.type: - raise HTTPException(status_code=400, detail="Missing activity type") + return HttpResponse(status=400, body={"detail": "Missing activity type"}) - async def logic(turn_context: TurnContext): + async def logic(turn_context: TurnContext) -> None: if not turn_context.activity.id: return - # Handle activity with botframework handler if self.handler: await self.handler.on_turn(turn_context) # Grab the auth header from the inbound request - auth_header = request.headers["Authorization"] if "Authorization" in request.headers else "" + auth_header = headers.get("authorization") or headers.get("Authorization") or "" await self.adapter.process_activity(auth_header, activity_bf, logic) - # Call parent's on_activity_request to handle Teams processing - return await super().on_activity_request(core_activity, request, response) + # Process through Teams pipeline + core_activity = CoreActivity.model_validate(body) + token = cast( + Any, + SimpleNamespace( + app_id="", + app_display_name="", + tenant_id="", + service_url=core_activity.service_url or "", + from_="azure", + from_id="", + is_expired=lambda: False, + ), + ) + + result = await self.on_activity_event(ActivityEvent(body=core_activity, token=token)) + + # Format response + status_code = 200 + resp_body: Any = None + if hasattr(result, "model_dump"): + resp_dict = result.model_dump(exclude_none=True) + elif isinstance(result, dict): + resp_dict = result + else: + resp_dict = {} + + if "status" in resp_dict: + status_code = resp_dict.get("status", 200) + if "body" in resp_dict: + resp_body = resp_dict.get("body") + + return HttpResponse(status=status_code, body=resp_body) - except HTTPException as http_err: - self.logger.error(f"HTTP error processing activity: {http_err}", exc_info=True) - raise except Exception as err: self.logger.error(f"Error processing activity: {err}", exc_info=True) - raise HTTPException(status_code=500, detail=str(err)) from err + return HttpResponse(status=500, body={"detail": str(err)}) diff --git a/packages/botbuilder/tests/test_botbuilder_plugin.py b/packages/botbuilder/tests/test_botbuilder_plugin.py index 0f604b10..477185aa 100644 --- a/packages/botbuilder/tests/test_botbuilder_plugin.py +++ b/packages/botbuilder/tests/test_botbuilder_plugin.py @@ -10,8 +10,9 @@ from botbuilder.core import ActivityHandler, TurnContext from botbuilder.integration.aiohttp import CloudAdapter from botbuilder.schema import Activity -from fastapi import HTTPException, Request, Response -from microsoft_teams.api import Credentials +from microsoft_teams.api import Credentials, InvokeResponse +from microsoft_teams.apps.http import HttpServer +from microsoft_teams.apps.http.adapter import HttpRequest from microsoft_teams.botbuilder import BotBuilderPlugin @@ -24,20 +25,26 @@ def mock_logger(self): @pytest.fixture def plugin_without_adapter(self): - plugin = BotBuilderPlugin(skip_auth=True) + plugin = BotBuilderPlugin() plugin.credentials = MagicMock(spec=Credentials) plugin.credentials.client_id = "abc" plugin.credentials.client_secret = "secret" plugin.credentials.tenant_id = "tenant-123" + plugin.server = MagicMock(spec=HttpServer) + plugin.logger = MagicMock() return plugin @pytest.fixture def plugin_with_adapter(self) -> BotBuilderPlugin: adapter = MagicMock(spec=CloudAdapter) - plugin = BotBuilderPlugin(adapter=adapter, skip_auth=True) - plugin._handle_activity_request = AsyncMock(return_value="fake_result") # pyright: ignore[reportPrivateUsage] + plugin = BotBuilderPlugin(adapter=adapter) handler = AsyncMock(spec=ActivityHandler) plugin.handler = handler + plugin.server = MagicMock(spec=HttpServer) + plugin.logger = MagicMock() + + # Set up the on_activity_event handler + plugin.on_activity_event = AsyncMock(return_value=InvokeResponse(status=200)) return plugin @pytest.mark.asyncio @@ -57,9 +64,15 @@ async def test_on_init_creates_adapter_when_missing(self, plugin_without_adapter mock_adapter_class.assert_called_once() assert plugin_without_adapter.adapter == "mock_adapter" + # Should have registered route via server + plugin_without_adapter.server.register_route.assert_called_once() + call_args = plugin_without_adapter.server.register_route.call_args + assert call_args[0][0] == "POST" + assert call_args[0][1] == "/api/messages" + @pytest.mark.asyncio - async def test_on_activity_request_calls_adapter_and_handler(self, plugin_with_adapter: BotBuilderPlugin): - # Mock request and response + async def test_handle_activity_calls_adapter_and_handler(self, plugin_with_adapter: BotBuilderPlugin): + """Test that _handle_activity calls adapter and handler.""" activity_data = { "type": "message", "id": "activity-id", @@ -68,27 +81,20 @@ async def test_on_activity_request_calls_adapter_and_handler(self, plugin_with_a "conversation": {"id": "conv1"}, "serviceUrl": "https://service.url", } - from microsoft_teams.apps.events import CoreActivity - - request = AsyncMock(spec=Request) - request.json.return_value = activity_data - request.headers = {"Authorization": "Bearer token"} - response = MagicMock(spec=Response) + request = HttpRequest( + body=activity_data, + headers={"Authorization": "Bearer token"}, + ) # Mock adapter.process_activity to call logic with a mock TurnContext - async def fake_process_activity(auth_header, activity, logic): # type: ignore - print("Inside fake_process_activity") + async def fake_process_activity(auth_header, activity, logic): await logic(MagicMock(spec=TurnContext)) assert plugin_with_adapter.adapter is not None - plugin_with_adapter.adapter.process_activity = AsyncMock(side_effect=fake_process_activity) - # Convert activity_data to CoreActivity - core_activity = CoreActivity(**activity_data) - - await plugin_with_adapter.on_activity_request(core_activity, request, response) + result = await plugin_with_adapter._handle_activity(request) # Ensure adapter.process_activity called with correct auth and activity plugin_with_adapter.adapter.process_activity.assert_called_once() @@ -97,27 +103,25 @@ async def fake_process_activity(auth_header, activity, logic): # type: ignore assert isinstance(called_activity, Activity) # Ensure handler called via TurnContext - plugin_with_adapter.handler.on_turn.assert_awaited() # type: ignore + plugin_with_adapter.handler.on_turn.assert_awaited() - @pytest.mark.asyncio - async def test_on_activity_request_raises_http_exception_on_adapter_error( - self, plugin_with_adapter: BotBuilderPlugin - ): - from microsoft_teams.apps.events import CoreActivity + # Should return a valid HttpResponse + assert result["status"] == 200 + @pytest.mark.asyncio + async def test_handle_activity_returns_error_on_adapter_error(self, plugin_with_adapter: BotBuilderPlugin): + """Test that _handle_activity returns 500 on adapter error.""" activity_data = {"type": "message", "id": "activity-id"} - request = AsyncMock(spec=Request) - request.json.return_value = activity_data - request.headers = {} - response = MagicMock(spec=Response) - assert plugin_with_adapter.adapter is not None + request = HttpRequest( + body=activity_data, + headers={}, + ) + assert plugin_with_adapter.adapter is not None plugin_with_adapter.adapter.process_activity = AsyncMock(side_effect=Exception("fail")) - # Convert activity_data to CoreActivity - core_activity = CoreActivity(**activity_data) + result = await plugin_with_adapter._handle_activity(request) - with pytest.raises(HTTPException) as exc: - await plugin_with_adapter.on_activity_request(core_activity, request, response) - assert exc.value.status_code == 500 + assert result["status"] == 500 + assert result["body"]["detail"] == "fail" diff --git a/packages/devtools/src/microsoft_teams/devtools/devtools_plugin.py b/packages/devtools/src/microsoft_teams/devtools/devtools_plugin.py index 5f8d6b70..3b2d0188 100644 --- a/packages/devtools/src/microsoft_teams/devtools/devtools_plugin.py +++ b/packages/devtools/src/microsoft_teams/devtools/devtools_plugin.py @@ -23,7 +23,6 @@ DependencyMetadata, ErrorEvent, EventMetadata, - HttpPlugin, LoggerDependencyOptions, Plugin, PluginActivityEvent, @@ -49,7 +48,6 @@ class DevToolsPlugin(PluginBase): logger: Annotated[Logger, LoggerDependencyOptions()] id: Annotated[Optional[TokenProtocol], DependencyMetadata(optional=True)] - http: Annotated[HttpPlugin, DependencyMetadata()] on_error_event: Annotated[Callable[[ErrorEvent], None], EventMetadata(name="error")] on_activity_event: Annotated[Callable[[ActivityEvent], None], EventMetadata(name="activity")] diff --git a/packages/mcpplugin/src/microsoft_teams/mcpplugin/server_plugin.py b/packages/mcpplugin/src/microsoft_teams/mcpplugin/server_plugin.py index f56bdf13..ea23d1d4 100644 --- a/packages/mcpplugin/src/microsoft_teams/mcpplugin/server_plugin.py +++ b/packages/mcpplugin/src/microsoft_teams/mcpplugin/server_plugin.py @@ -13,7 +13,8 @@ from microsoft_teams.ai import Function, FunctionHandler from microsoft_teams.apps import ( DependencyMetadata, - HttpPlugin, + FastAPIAdapter, + HttpServer, Plugin, PluginBase, PluginStartEvent, @@ -40,7 +41,7 @@ class McpServerPlugin(PluginBase): """ # Dependency injection - http: Annotated[HttpPlugin, DependencyMetadata()] + http_server: Annotated[HttpServer, DependencyMetadata()] def __init__(self, name: str = "teams-mcp-server", path: str = "/mcp", logger: logging.Logger | None = None): """ @@ -158,10 +159,13 @@ async def on_start(self, event: PluginStartEvent) -> None: return try: + adapter = self.http_server.adapter + if not isinstance(adapter, FastAPIAdapter): + raise RuntimeError("McpServerPlugin requires FastAPIAdapter. Custom adapters are not supported.") + # We mount the mcp server as a separate app at self.path mcp_http_app = self.mcp_server.http_app(path=self.path, transport="http") - self.http.lifespans.append(mcp_http_app.lifespan) - self.http.app.mount("/", mcp_http_app) + adapter.app.mount("/", mcp_http_app) self._mounted = True diff --git a/uv.lock b/uv.lock index 09da5283..1bbbec0a 100644 --- a/uv.lock +++ b/uv.lock @@ -15,6 +15,7 @@ members = [ "dialogs", "echo", "graph", + "http-adapters", "mcp-client", "mcp-server", "meetings", @@ -1172,6 +1173,29 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/07/c6/80c95b1b2b94682a72cbdbfb85b81ae2daffa4291fbfa1b1464502ede10d/hpack-4.1.0-py3-none-any.whl", hash = "sha256:157ac792668d995c657d93111f46b4535ed114f0c9c8d672271bbec7eae1b496", size = 34357, upload-time = "2025-01-22T21:44:56.92Z" }, ] +[[package]] +name = "http-adapters" +version = "0.1.0" +source = { virtual = "examples/http-adapters" } +dependencies = [ + { name = "dotenv" }, + { name = "httptools" }, + { name = "microsoft-teams-api" }, + { name = "microsoft-teams-apps" }, + { name = "starlette" }, + { name = "uvicorn" }, +] + +[package.metadata] +requires-dist = [ + { name = "dotenv", specifier = ">=0.9.9" }, + { name = "httptools" }, + { name = "microsoft-teams-api", editable = "packages/api" }, + { name = "microsoft-teams-apps", editable = "packages/apps" }, + { name = "starlette" }, + { name = "uvicorn" }, +] + [[package]] name = "httpcore" version = "1.0.9"