Replace HttpPlugin with HttpServer + HttpServerAdapter

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 }
+```

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.14"
+dependencies = [
+    "dotenv>=0.9.9",
+    "microsoft-teams-apps",
+    "microsoft-teams-api",
+    "starlette",
+    "uvicorn",
+    "httptools",
+]
+
+[tool.uv.sources]
+microsoft-teams-apps = { workspace = true }

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("""
+        <h1>FastAPI + Teams Bot</h1>
+        <p>Your FastAPI server is running with a Teams bot!</p>
+        <ul>
+            <li><a href="/health">Health Check</a></li>
+            <li><a href="/api/users">API: Users</a></li>
+            <li><b>/api/messages</b> — Teams bot endpoint (added by SDK)</li>
+        </ul>
+    """)
+
+
+# 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())

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

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())

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)

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__)