Add host-owned ASGI route registration

examples/a2a/src/main.py

@@ -80,6 +80,11 @@ async def handle_message(ctx: ActivityContext[MessageActivity]) -> None:
         await ctx.reply(reply)
 
 
+# Register Teams routes synchronously so the host-owned FastAPI route table is
+# complete before the server starts.
+app.register_routes()
+
+
 async def main() -> None:
     # Build this bot's own AgentCard — equivalent to AgentCardFactory.Build(config) in C#.
     agent_card = build_agent_card(config)
@@ -89,7 +94,7 @@ async def main() -> None:
     a2a_starlette = make_a2a_app(teams_app=app, agent=bot_agent, config=config, agent_card=agent_card)
     fastapi_app.mount("/a2a", a2a_starlette.build())
 
-    await app.initialize()
+    await app.start_plugins()
 
     port = int(getenv("PORT", "3978"))
     server = uvicorn.Server(uvicorn.Config(fastapi_app, host="0.0.0.0", port=port, log_level="info"))

examples/http-adapters/README.md

@@ -1,12 +1,12 @@
 # HTTP Adapters Examples
 
-Examples showing how to use custom `HttpServerAdapter` implementations and non-managed server patterns with the Teams Python SDK.
+Examples showing how to use shipped `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.
+The shipped `StarletteAdapter` implementation for [Starlette](https://www.starlette.io/).
 
 **Pattern**: Custom adapter, SDK-managed server lifecycle (`app.start()`)
 
@@ -18,7 +18,7 @@ python src/starlette_echo.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`)
+**Pattern**: Default `FastAPIAdapter` with user-provided FastAPI instance, user-managed server (`app.register_routes()` + `await app.start_plugins()` + your own `uvicorn.Server`)
 
 ```bash
 python src/fastapi_non_managed.py
@@ -30,10 +30,14 @@ python src/fastapi_non_managed.py
 
 | | Managed | Non-Managed |
 |---|---|---|
-| **Entry point** | `app.start(port)` | `app.initialize()` + start server yourself |
+| **Entry point** | `app.start(port)` | `app.register_routes()` + `await app.start_plugins()` + start server yourself |
 | **Who starts the server** | The SDK (via adapter) | You |
 | **When to use** | New apps, simple setup | Existing apps, custom server config |
 
+`app.register_routes()` is synchronous and registers the Teams messaging route
+without running async plugin initialization. Run `await app.start_plugins()` from
+your host's startup hook when the event loop is available.
+
 ### Writing a Custom Adapter
 
 Implement the `HttpServerAdapter` protocol:

examples/http-adapters/pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "http-adapters"
 version = "0.1.0"
-description = "Examples showing custom HttpServerAdapter and non-managed server patterns"
+description = "Examples showing HttpServerAdapter and non-managed server patterns"
 readme = "README.md"
 requires-python = ">=3.11,<4.0"
 dependencies = [

examples/http-adapters/src/fastapi_non_managed.py

@@ -8,8 +8,8 @@
 # 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.
+# with your own routes, wrap it in a FastAPIAdapter, register Teams routes
+# synchronously, then start uvicorn yourself.
 #
 # This is ideal when:
 # - You have an existing FastAPI app and want to add Teams bot support
@@ -81,20 +81,24 @@ async def handle_message(ctx: ActivityContext[MessageActivity]):
     await ctx.send(f"[FastAPI non-managed] You said: '{ctx.activity.text}'")
 
 
+# 6. Register routes synchronously — registers /api/messages on our FastAPI app.
+#    Does NOT start a server and does NOT run async plugin init hooks.
+app.register_routes()
+
+
 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()
+    # 7. Initialize async plugins during host startup.
+    await app.start_plugins()
 
     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
+    # 8. 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()

examples/http-adapters/src/starlette_echo.py

@@ -5,7 +5,7 @@
 
 # Starlette Echo Bot
 # ==================
-# Teams echo bot using a custom StarletteAdapter.
+# Teams echo bot using the shipped 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.
@@ -17,10 +17,9 @@
 import re
 
 from microsoft_teams.api import MessageActivity
-from microsoft_teams.apps import ActivityContext, App
+from microsoft_teams.apps import ActivityContext, App, StarletteAdapter
 from starlette.requests import Request
 from starlette.responses import JSONResponse
-from starlette_adapter import StarletteAdapter
 
 # 1. Create adapter
 adapter = StarletteAdapter()

examples/mcp-server/src/main.py

@@ -16,10 +16,10 @@
 
 
 async def main() -> None:
-    # app.initialize() must be called before mounting the MCP app so that
+    # app.register_routes() must be called before mounting the MCP app so that
     # /api/messages is registered first — FastAPI routes take priority over
     # mounted sub-applications, and the MCP mount uses a catch-all path (/).
-    await app.initialize()
+    app.register_routes()
 
     adapter = app.server.adapter
     if not isinstance(adapter, FastAPIAdapter):

packages/apps/README.md

@@ -39,6 +39,37 @@ async def handle_message(ctx: ActivityContext[MessageActivity]):
 await app.start()
 ```
 
+## Host-Owned ASGI Lifecycle
+
+If another ASGI app owns the server lifecycle, register Teams routes synchronously
+and initialize plugins during startup:
+
+```python
+from fastapi import FastAPI
+from microsoft_teams.apps import App, FastAPIAdapter
+
+asgi_app = FastAPI()
+teams = App(http_server_adapter=FastAPIAdapter(app=asgi_app))
+
+routes = teams.register_routes()  # registers POST /api/messages, does not start a server
+
+
+async def startup():
+    # Call this from your host's ASGI startup hook.
+    await teams.start_plugins()
+```
+
+For Starlette hosts, use the shipped `StarletteAdapter`:
+
+```python
+from microsoft_teams.apps import App, StarletteAdapter
+from starlette.applications import Starlette
+
+asgi_app = Starlette()
+teams = App(http_server_adapter=StarletteAdapter(app=asgi_app))
+teams.register_routes()
+```
+
 ## OAuth and Graph Integration
 
 ```python

packages/apps/src/microsoft_teams/apps/__init__.py

@@ -10,7 +10,7 @@
 from .auth import *  # noqa: F403
 from .contexts import *  # noqa: F403
 from .events import *  # noqa: F401, F403
-from .http import FastAPIAdapter, HttpServer, HttpServerAdapter
+from .http import FastAPIAdapter, HttpRoute, HttpServer, HttpServerAdapter, StarletteAdapter
 from .http_stream import HttpStream
 from .options import AppOptions
 from .plugins import *  # noqa: F401, F403
@@ -26,6 +26,8 @@
     "HttpServer",
     "HttpServerAdapter",
     "FastAPIAdapter",
+    "StarletteAdapter",
+    "HttpRoute",
     "HttpStream",
     "ActivityContext",
     "to_threaded_conversation_id",

packages/apps/src/microsoft_teams/apps/app.py

@@ -52,7 +52,7 @@
     get_event_type_from_signature,
     is_registered_event,
 )
-from .http import FastAPIAdapter, HttpServer
+from .http import FastAPIAdapter, HttpRoute, HttpServer
 from .http.adapter import HttpRequest, HttpResponse
 from .options import AppOptions, InternalAppOptions
 from .plugins import PluginBase, PluginStartEvent
@@ -123,6 +123,8 @@ def __init__(self, **options: Unpack[AppOptions]):
         self.container.set_provider("HttpServer", providers.Object(self.server))
 
         self._port: Optional[int] = None
+        self._routes_registered = False
+        self._plugins_initialized = False
         self._initialized = False
 
         # initialize ActivitySender for sending activities
@@ -187,6 +189,36 @@ def id(self) -> Optional[str]:
             return None
         return self.credentials.client_id
 
+    def register_routes(self) -> list[HttpRoute]:
+        """
+        Register the Teams messaging endpoint synchronously.
+
+        This wires the app activity dispatcher, configures inbound auth, and
+        registers the messaging route on the configured HTTP adapter. It does
+        not run plugin ``on_init`` hooks and does not start or stop a server.
+        """
+        try:
+            return self._register_routes()
+        except Exception as error:
+            logger.error(f"Failed to register app routes: {error}")
+            self._events.emit("error", ErrorEvent(error, context={"method": "register_routes"}))
+            raise
+
+    async def start_plugins(self) -> None:
+        """
+        Run plugin ``on_init`` hooks without registering routes or starting a server.
+
+        This is intended for host-owned ASGI lifecycles that call
+        :meth:`register_routes` synchronously during route contribution, then
+        initialize async plugins during application startup.
+        """
+        try:
+            await self._initialize_plugins()
+        except Exception as error:
+            logger.error(f"Failed to initialize plugins: {error}")
+            self._events.emit("error", ErrorEvent(error, context={"method": "start_plugins"}))
+            raise
+
     async def initialize(self) -> None:
         """
         Initialize the Teams application without starting the HTTP server.
@@ -200,20 +232,11 @@ async def initialize(self) -> None:
 
         try:
             # Initialize plugins first (they may register routes, e.g. BotBuilder's /api/messages)
-            for plugin in self.plugins:
-                self._plugin_processor.inject(plugin)
-                if hasattr(plugin, "on_init") and callable(plugin.on_init):
-                    await plugin.on_init()
+            await self._initialize_plugins()
 
             # Initialize HttpServer (JWT validation + messaging endpoint route)
-            self.server.on_request = self._process_activity_event
-            self.server.initialize(
-                credentials=self.credentials,
-                skip_auth=self.options.skip_auth,
-                cloud=self.cloud,
-            )
+            self._register_routes()
 
-            self._initialized = True
             logger.info("Teams app initialized successfully")
 
         except Exception as error:
@@ -299,7 +322,10 @@ async def send(self, conversation_id: str, activity: str | ActivityParams | Adap
         """
 
         if not self._initialized:
-            raise ValueError("app not initialized - call app.initialize() or app.start() first")
+            raise ValueError(
+                "app not initialized - call app.initialize(), app.start(), "
+                "or app.register_routes() followed by app.start_plugins() first"
+            )
 
         if self.id is None:
             raise ValueError("app credentials not configured")
@@ -593,6 +619,32 @@ async def _stop_plugins(self) -> None:
             if hasattr(plugin, "on_stop") and callable(plugin.on_stop):
                 await plugin.on_stop()
 
+    async def _initialize_plugins(self) -> None:
+        if self._plugins_initialized:
+            return
+
+        for plugin in self.plugins:
+            self._plugin_processor.inject(plugin)
+            if hasattr(plugin, "on_init") and callable(plugin.on_init):
+                await plugin.on_init()
+
+        self._plugins_initialized = True
+        self._refresh_initialized()
+
+    def _register_routes(self) -> list[HttpRoute]:
+        self.server.on_request = self._process_activity_event
+        routes = self.server.initialize(
+            credentials=self.credentials,
+            skip_auth=self.options.skip_auth,
+            cloud=self.cloud,
+        )
+        self._routes_registered = True
+        self._refresh_initialized()
+        return routes
+
+    def _refresh_initialized(self) -> None:
+        self._initialized = self._routes_registered and self._plugins_initialized
+
     async def _get_bot_token(self):
         return await self._token_manager.get_bot_token()
 

packages/apps/src/microsoft_teams/apps/http/__init__.py

@@ -3,16 +3,19 @@
 Licensed under the MIT License.
 """
 
-from .adapter import HttpMethod, HttpRequest, HttpResponse, HttpRouteHandler, HttpServerAdapter
+from .adapter import HttpMethod, HttpRequest, HttpResponse, HttpRoute, HttpRouteHandler, HttpServerAdapter
 from .fastapi_adapter import FastAPIAdapter
 from .http_server import HttpServer
+from .starlette_adapter import StarletteAdapter
 
 __all__ = [
     "HttpMethod",
     "HttpRequest",
     "HttpResponse",
+    "HttpRoute",
     "HttpRouteHandler",
     "HttpServer",
     "HttpServerAdapter",
     "FastAPIAdapter",
+    "StarletteAdapter",
 ]

packages/apps/src/microsoft_teams/apps/http/adapter.py

@@ -3,7 +3,7 @@
 Licensed under the MIT License.
 """
 
-from typing import Dict, Literal, Protocol, TypedDict, runtime_checkable
+from typing import Dict, Literal, NamedTuple, Protocol, TypedDict, runtime_checkable
 
 HttpMethod = Literal["POST"]
 
@@ -22,6 +22,14 @@ class HttpRouteHandler(Protocol):
     async def __call__(self, request: HttpRequest) -> HttpResponse: ...
 
 
+class HttpRoute(NamedTuple):
+    """Framework-agnostic HTTP route registered by the Teams app."""
+
+    method: HttpMethod
+    path: str
+    handler: HttpRouteHandler
+
+
 @runtime_checkable
 class HttpServerAdapter(Protocol):
     """Protocol for framework-specific HTTP server adapters.