Repository: getappmap/appmap-python (language: python)

Issue:
FastAPI support
appmap-python will record AppMaps of FastAPI code via test case recording.

Enhance appmap-python to support requests recording and remote recording for FastAPI: https://fastapi.tiangolo.com/

Ensure that the output includes HTTP server requests and responses.


Discussion / hints:
Per @apotterri  FastAPI is built on [Starlette](https://www.starlette.io/), HTTP framework that implements [ASGI](https://asgi.readthedocs.io/en/latest/index.html). The 'A' in ASGI stands for "Asynchronous", and there's no requirement that there be a single thread that processes an entire request.

Adding support for it is almost certainly going to be more complicated than just hooking a couple of framework methods.
@apotterri 
took some inspiration from sentry_sdk

```
import asyncio
from functools import wraps
import threading
import time

import fastapi

from _appmap.env import Env
from _appmap.metadata import Metadata

logger = Env.current.getLogger(__name__)


try:
    from gevent import get_hub as get_gevent_hub  # type: ignore
    from gevent.monkey import get_original, is_module_patched  # type: ignore
    from gevent.threadpool import ThreadPool  # type: ignore

    thread_sleep = get_original("time", "sleep")
except ImportError:

    def get_gevent_hub():
        return None

    thread_sleep = time.sleep

    def is_module_patched(*args, **kwargs):
        # unable to import from gevent means no modules have been patched
        return False

    ThreadPool = None


def is_gevent():
    return is_module_patched("threading") or is_module_patched("_thread")


def get_current_thread_id(thread=None):
    """
    Try to get the id of the current thread, with various fall backs.
    """

    # if a thread is specified, that takes priority
    if thread is not None:
        try:
            thread_id = thread.ident
            if thread_id is not None:
                return thread_id
        except AttributeError:
            pass

    # if the app is using gevent, we should look at the gevent hub first
    # as the id there differs from what the threading module reports
    if is_gevent():
        gevent_hub = get_gevent_hub()
        if gevent_hub is not None:
            try:
                # this is undocumented, so wrap it in try except to be safe
                return gevent_hub.thread_ident
            except AttributeError:
                pass

    # use the current thread's id if possible
    try:
        current_thread_id = threading.current_thread().ident
        if current_thread_id is not None:
            return current_thread_id
    except AttributeError:
        pass

    # if we can't get the current thread id, fall back to the main thread id
    try:
        main_thread_id = threading.main_thread().ident
        if main_thread_id is not None:
            return main_thread_id
    except AttributeError:
        pass

    # we've tried everything, time to give up
    return None


def patch_get_request_handler():
    old_get_request_handler = fastapi.routing.get_request_handler

    def _appmap_get_request_handler(*args, **kwargs):
        dependant = kwargs.get("dependant")
        if (
            dependant
            and dependant.call is not None
            and not asyncio.iscoroutinefunction(dependant.call)
        ):
            old_call = dependant.call

            @wraps(old_call)
            def _appmap_call(*args, **kwargs):
                print('-' * 50)
                print('_appmap_call')

                print('get_current_thread_id', get_current_thread_id())
                return old_call(*args, **kwargs)

            dependant.call = _appmap_call

        old_app = old_get_request_handler(*args, **kwargs)

        async def _appmap_app(*args, **kwargs):
            Metadata.add_framework("fastapi", fastapi.__version__)

            print('-' * 50)
            print('_appmap_app')

            request = args[0]
            print(request)

            response = await old_app(*args, **kwargs)
            print(response)

            return response

        return _appmap_app

    fastapi.routing.get_request_handler = _appmap_get_request_handler


if Env.current.enabled:
    patch_get_request_handler()

```

Please let me know if you think this might work.

I haven't had a chance to go through `MiddlewareInserter` and `AppmapMiddleware`, so don't really understand how to use it yet here. `fastapi.routing.get_request_handler` gets called on every request.
Another thing to note is, appmap still supports Python 3.7.2, but FastAPI requires >=3.8
@apotterri @kgilpin FastAPI supports middlewares, but injecting them like flask and django doesn't seem to be possible, SO FAR.

Since FastAPI is run using uvicorn, there is no common fixed entry point inside Starlette or FastAPI. Need to explore if we can somehow inject directly inside uvicorn.

Else, we can create middleware and would need the dev to do something like
```
app = FastAPI()
app.add_middleware(AppmapFastAPIMiddleware)
```

https://semaphoreci.com/blog/custom-middleware-fastapi
https://stackoverflow.com/questions/71525132/how-to-write-a-custom-fastapi-middleware-class

@virajkanwade welcome back, and thanks a lot for digging into this!

I think it would be better to implement an ASGI middleware to manage request recording, rather than tying ourselves to FastAPI specifically. This would mean that we'd work with any ASGI server, and so also support all the frameworks that run on them. It might also mean that we won't be limited by FastAPI's 3.8 requirement (though that will depend on how we test the new middleware, I guess).

Implementing the middleware itself should be pretty straightforward, as it will look a lot like `AppmapMiddleware` (and they may even be able to share some code).

Maybe somewhat trickier will be changing the way the current `Recorder` is managed. Currently, it's stored in a thread-local variable: https://github.com/getappmap/appmap-python/blob/2c2025b676f3a44bb933935aee927e3a69746071/_appmap/utils.py#L64 . It seems like changing the implementation to use a `ContextVar` instead of a `threading.local` will be sufficient, but I'm not 100% sure. 

We should definitely see if we can hook something in uvicorn (and any other support we want to support) to inject the new middleware. The less the user has to do to get AppMaps, the better. 
@virajkanwade thanks for diving into this! FastAPI would be an awesome enhancement.
@apotterri uvicorn injection

```
from typing import List, Optional, Tuple, Union, cast

import uvicorn
from uvicorn._types import (
    ASGI3Application,
    ASGIReceiveCallable,
    ASGISendCallable,
    ASGISendEvent,
    Scope,
)
from uvicorn.config import Config

from _appmap import wrapt
from _appmap.env import Env
from _appmap.web_framework import AppmapMiddleware, MiddlewareInserter


class AppmapUvicornMiddleware:
    def __init__(
        self,
        app: "ASGI3Application",
    ) -> None:
        self.app = app

    async def __call__(
        self, scope: "Scope", receive: "ASGIReceiveCallable", send: "ASGISendCallable"
    ) -> None:
        # scope["type"]
        #   "http": for HTTP requests.
        #   "websocket": for WebSocket connections.
        #   "lifespan": for ASGI lifespan messages.

        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        client_addr: Optional[Tuple[str, int]] = scope.get("client")
        client_host = client_addr[0] if client_addr else None

        print("client_host", client_host)
        print("scheme", scope["scheme"])
        print("method", scope["method"])
        print("path", scope["path"])
        print("query_string", scope["query_string"])
        print("headers", scope["headers"])

        print(scope)

        async def send_wrapper(message: ASGISendEvent) -> None:
            print(message)

            await send(message)

        return await self.app(scope, receive, send_wrapper)


def install_extension(wrapped, self_obj, args, kwargs):
    wrapped(*args, **kwargs)
    app = self_obj.loaded_app
    if app:
        app = AppmapUvicornMiddleware(app)

        self_obj.loaded_app = app


if Env.current.enabled:
    Config.load = wrapt.wrap_function_wrapper("uvicorn.config", "Config.load", install_extension)
```
Nice! Thank for this.



Fix patch:
diff --git a/_appmap/configuration.py b/_appmap/configuration.py
index fed32b07..bbfc8324 100644
--- a/_appmap/configuration.py
+++ b/_appmap/configuration.py
@@ -2,6 +2,7 @@
 Manage Configuration AppMap recorder for Python.
 """
 
+import importlib.metadata
 import inspect
 import os
 import sys
@@ -9,7 +10,6 @@
 from pathlib import Path
 from textwrap import dedent
 
-import importlib_metadata
 import yaml
 from yaml.parser import ParserError
 
@@ -142,8 +142,6 @@ def __init__(self):
 
         self._load_config()
         self._load_functions()
-        logger.info("config: %s", self._config)
-        logger.debug("package_functions: %s", self.package_functions)
 
         if "labels" in self._config:
             self.labels.append(self._config["labels"])
@@ -314,7 +312,7 @@ class DistMatcher(PathMatcher):
     def __init__(self, dist, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.dist = dist
-        self.files = [str(pp.locate()) for pp in importlib_metadata.files(dist)]
+        self.files = [str(pp.locate()) for pp in importlib.metadata.files(dist)]
 
     def matches(self, filterable):
         try:
@@ -415,3 +413,8 @@ def initialize():
 
 
 initialize()
+
+c = Config()
+logger.info("config: %s", c._config)
+logger.debug("package_functions: %s", c.package_functions)
+logger.info("env: %r", os.environ)
diff --git a/_appmap/env.py b/_appmap/env.py
index 8a558c88..a4ffd719 100644
--- a/_appmap/env.py
+++ b/_appmap/env.py
@@ -4,6 +4,7 @@
 import logging.config
 import os
 from contextlib import contextmanager
+from datetime import datetime
 from os import environ
 from pathlib import Path
 from typing import cast
@@ -124,11 +125,10 @@ def getLogger(self, name) -> trace_logger.TraceLogger:
     def _configure_logging(self):
         trace_logger.install()
 
-        log_level = self.get("APPMAP_LOG_LEVEL", "warning").upper()
-
+        log_level = self.get("APPMAP_LOG_LEVEL", "warn").upper()
+        disable_log = os.environ.get("APPMAP_DISABLE_LOG_FILE", "false").upper() != "FALSE"
         log_config = self.get("APPMAP_LOG_CONFIG")
-        log_stream = self.get("APPMAP_LOG_STREAM", "stderr")
-        log_stream = "ext://sys.%s" % (log_stream)
+        now = datetime.now()
         config_dict = {
             "version": 1,
             "disable_existing_loggers": False,
@@ -138,9 +138,7 @@ def _configure_logging(self):
                     "format": "[{asctime}] {levelname} {name}: {message}",
                 }
             },
-            "handlers": {
-                "default": {"class": "logging.StreamHandler", "formatter": "default"}
-            },
+            "handlers": {"default": {"class": "logging.StreamHandler", "formatter": "default"}},
             "loggers": {
                 "appmap": {
                     "level": log_level,
@@ -154,6 +152,20 @@ def _configure_logging(self):
                 },
             },
         }
+        if not disable_log:
+            # Default to being more verbose if we're logging to a file, but
+            # still allow the level to be overridden.
+            log_level = self.get("APPMAP_LOG_LEVEL", "info").upper()
+            loggers = config_dict["loggers"]
+            loggers["appmap"]["level"] = loggers["_appmap"]["level"] = log_level
+            config_dict["handlers"] = {
+                "default": {
+                    "class": "logging.FileHandler",
+                    "formatter": "default",
+                    "filename": f"appmap-{now:%Y%m%d%H%M%S}-{os.getpid()}.log",
+                }
+            }
+
         if log_config is not None:
             name, level = log_config.split("=", 2)
             config_dict["loggers"].update(
diff --git a/_appmap/event.py b/_appmap/event.py
index 76385ed6..0b1964c1 100644
--- a/_appmap/event.py
+++ b/_appmap/event.py
@@ -10,11 +10,11 @@
 from .recorder import Recorder
 from .utils import (
     FnType,
+    FqFnName,
     appmap_tls,
     compact_dict,
     fqname,
     get_function_location,
-    split_function_name,
 )
 
 logger = Env.current.getLogger(__name__)
@@ -173,7 +173,7 @@ def to_dict(self, value):
 
 
 class CallEvent(Event):
-    __slots__ = ["_fn", "static", "receiver", "parameters", "labels"]
+    __slots__ = ["_fn", "_fqfn", "static", "receiver", "parameters", "labels"]
 
     @staticmethod
     def make(fn, fntype):
@@ -209,10 +209,9 @@ def make_params(filterable):
             # going to log a message about a mismatch.
             wrapped_sig = inspect.signature(fn, follow_wrapped=True)
             if sig != wrapped_sig:
-                logger.debug(
-                    "signature of wrapper %s.%s doesn't match wrapped",
-                    *split_function_name(fn)
-                )
+                logger.debug("signature of wrapper %r doesn't match wrapped", fn)
+                logger.debug("sig: %r", sig)
+                logger.debug("wrapped_sig: %r", wrapped_sig)
 
         return [Param(p) for p in sig.parameters.values()]
 
@@ -270,17 +269,17 @@ def set_params(params, instance, args, kwargs):
     @property
     @lru_cache(maxsize=None)
     def function_name(self):
-        return split_function_name(self._fn)
+        return self._fqfn.fqfn
 
     @property
     @lru_cache(maxsize=None)
     def defined_class(self):
-        return self.function_name[0]
+        return self._fqfn.fqclass
 
     @property
     @lru_cache(maxsize=None)
     def method_id(self):
-        return self.function_name[1]
+        return self._fqfn.fqfn[1]
 
     @property
     @lru_cache(maxsize=None)
@@ -308,6 +307,7 @@ def comment(self):
     def __init__(self, fn, fntype, parameters, labels):
         super().__init__("call")
         self._fn = fn
+        self._fqfn = FqFnName(fn)
         self.static = fntype in FnType.STATIC | FnType.CLASS | FnType.MODULE
         self.receiver = None
         if fntype in FnType.CLASS | FnType.INSTANCE:
@@ -351,7 +351,15 @@ class MessageEvent(Event):  # pylint: disable=too-few-public-methods
     def __init__(self, message_parameters):
         super().__init__("call")
         self.message = []
-        for name, value in message_parameters.items():
+        self.message_parameters = message_parameters
+
+    @property
+    def message_parameters(self):
+        return self.message
+
+    @message_parameters.setter
+    def message_parameters(self, params):
+        for name, value in params.items():
             message_object = describe_value(name, value)
             self.message.append(message_object)
 
@@ -386,6 +394,7 @@ def __init__(self, request_method, url, message_parameters, headers=None):
 
 
 # pylint: disable=too-few-public-methods
+_NORMALIZED_PATH_INFO_ATTR = "normalized_path_info"
 class HttpServerRequestEvent(MessageEvent):
     """A call AppMap event representing an HTTP server request."""
 
@@ -406,7 +415,7 @@ def __init__(
             "request_method": request_method,
             "protocol": protocol,
             "path_info": path_info,
-            "normalized_path_info": normalized_path_info,
+            _NORMALIZED_PATH_INFO_ATTR: normalized_path_info,
         }
 
         if headers is not None:
@@ -420,6 +429,14 @@ def __init__(
 
         self.http_server_request = compact_dict(request)
 
+    @property
+    def normalized_path_info(self):
+        return self.http_server_request.get(_NORMALIZED_PATH_INFO_ATTR, None)
+
+    @normalized_path_info.setter
+    def normalized_path_info(self, npi):
+        self.http_server_request[_NORMALIZED_PATH_INFO_ATTR] = npi
+
 
 class ReturnEvent(Event):
     __slots__ = ["parent_id", "elapsed"]
diff --git a/_appmap/fastapi.py b/_appmap/fastapi.py
new file mode 100644
index 00000000..6ffe0f54
--- /dev/null
+++ b/_appmap/fastapi.py
@@ -0,0 +1,32 @@
+"""
+This file contains a FastAPI app that is mounted on /_appmap to expose the remote-recording endpoint
+in a user's app.
+
+It should only be imported if other code has already verified that FastAPI is available.
+"""
+
+from fastapi import FastAPI, Response
+
+from . import remote_recording
+
+app = FastAPI()
+
+
+def _rr_response(fn):
+    body, rrstatus = fn()
+    return Response(content=body, status_code=rrstatus, media_type="application/json")
+
+
+@app.get("/record")
+def status():
+    return _rr_response(remote_recording.status)
+
+
+@app.post("/record")
+def start():
+    return _rr_response(remote_recording.start)
+
+
+@app.delete("/record")
+def stop():
+    return _rr_response(remote_recording.stop)
diff --git a/_appmap/importer.py b/_appmap/importer.py
index 5cbee04e..b853c384 100644
--- a/_appmap/importer.py
+++ b/_appmap/importer.py
@@ -10,12 +10,12 @@
 from _appmap import wrapt
 
 from .env import Env
-from .utils import FnType
+from .utils import FnType, Scope
 
 logger = Env.current.getLogger(__name__)
 
 
-Filterable = namedtuple("Filterable", "fqname obj")
+Filterable = namedtuple("Filterable", "scope fqname obj")
 
 
 class FilterableMod(Filterable):
@@ -23,10 +23,7 @@ class FilterableMod(Filterable):
 
     def __new__(cls, mod):
         fqname = mod.__name__
-        return super(FilterableMod, cls).__new__(cls, fqname, mod)
-
-    def classify_fn(self, _):
-        return FnType.MODULE
+        return super(FilterableMod, cls).__new__(cls, Scope.MODULE, fqname, mod)
 
 
 class FilterableCls(Filterable):
@@ -34,32 +31,28 @@ class FilterableCls(Filterable):
 
     def __new__(cls, clazz):
         fqname = "%s.%s" % (clazz.__module__, clazz.__qualname__)
-        return super(FilterableCls, cls).__new__(cls, fqname, clazz)
-
-    def classify_fn(self, static_fn):
-        return FnType.classify(static_fn)
+        return super(FilterableCls, cls).__new__(cls, Scope.CLASS, fqname, clazz)
 
 
 class FilterableFn(
     namedtuple(
         "FilterableFn",
-        Filterable._fields
-        + (
-            "scope",
-            "static_fn",
-        ),
+        Filterable._fields + ("static_fn",),
     )
 ):
     __slots__ = ()
 
     def __new__(cls, scope, fn, static_fn):
         fqname = "%s.%s" % (scope.fqname, fn.__name__)
-        self = super(FilterableFn, cls).__new__(cls, fqname, fn, scope, static_fn)
+        self = super(FilterableFn, cls).__new__(cls, scope.scope, fqname, fn, static_fn)
         return self
 
     @property
     def fntype(self):
-        return self.scope.classify_fn(self.static_fn)
+        if self.scope == Scope.MODULE:
+            return FnType.MODULE
+
+        return FnType.classify(self.static_fn)
 
 
 class Filter(ABC):  # pylint: disable=too-few-public-methods
@@ -161,6 +154,17 @@ def initialize(cls):
     def use_filter(cls, filter_class):
         cls.filter_stack.append(filter_class)
 
+    @classmethod
+    def instrument_function(cls, fn_name, filterableFn: FilterableFn, selected_functions=None):
+        # Only instrument the function if it was specifically called out for the package
+        # (e.g. because it should be labeled), or it's included by the filters
+        matched = cls.filter_chain.filter(filterableFn)
+        selected = selected_functions and fn_name in selected_functions
+        if selected or matched:
+            return cls.filter_chain.wrap(filterableFn)
+
+        return filterableFn.obj
+
     @classmethod
     def do_import(cls, *args, **kwargs):
         mod = args[0]
@@ -177,15 +181,10 @@ def instrument_functions(filterable, selected_functions=None):
             logger.trace("  functions %s", functions)
 
             for fn_name, static_fn, fn in functions:
-                # Only instrument the function if it was specifically called out for the package
-                # (e.g. because it should be labeled), or it's included by the filters
                 filterableFn = FilterableFn(filterable, fn, static_fn)
-                matched = cls.filter_chain.filter(filterableFn)
-                selected = selected_functions and fn_name in selected_functions
-                if selected or matched:
-                    new_fn = cls.filter_chain.wrap(filterableFn)
-                    if fn != new_fn:
-                        wrapt.wrap_function_wrapper(filterable.obj, fn_name, new_fn)
+                new_fn = cls.instrument_function(fn_name, filterableFn, selected_functions)
+                if new_fn != fn:
+                    wrapt.wrap_function_wrapper(filterable.obj, fn_name, new_fn)
 
         # Import Config here, to avoid circular top-level imports.
         from .configuration import Config  # pylint: disable=import-outside-toplevel
diff --git a/_appmap/metadata.py b/_appmap/metadata.py
index 9161f3e0..03ae0072 100644
--- a/_appmap/metadata.py
+++ b/_appmap/metadata.py
@@ -1,7 +1,6 @@
 """Shared metadata gathering"""
 
 import platform
-import re
 from functools import lru_cache
 
 from . import utils
diff --git a/_appmap/remote_recording.py b/_appmap/remote_recording.py
index e17f1088..218d2063 100644
--- a/_appmap/remote_recording.py
+++ b/_appmap/remote_recording.py
@@ -1,4 +1,5 @@
-""" remote_recording is a Flask app that can be mounted to expose the remote-recording endpoint. """
+""" remote_recording contains the functions neccessary to implement a remote-recording endpoint. """
+
 import json
 from threading import Lock
 
diff --git a/_appmap/utils.py b/_appmap/utils.py
index 69dcf600..cf321c66 100644
--- a/_appmap/utils.py
+++ b/_appmap/utils.py
@@ -1,17 +1,22 @@
-from contextvars import ContextVar
 import inspect
 import os
 import re
 import shlex
 import subprocess
-import threading
 import types
-from collections.abc import MutableMapping
-from enum import IntFlag, auto
+from contextlib import contextmanager
+from contextvars import ContextVar
+from enum import Enum, IntFlag, auto
+from typing import Any, Callable
 
 from .env import Env
 
 
+class Scope(Enum):
+    MODULE = 0
+    CLASS = 1
+
+
 def compact_dict(dictionary):
     """Return a copy of dictionary with None values filtered out."""
     return {k: v for k, v in dictionary.items() if v is not None}
@@ -41,27 +46,6 @@ def classify(fn):
         return FnType.INSTANCE
 
 
-class ThreadLocalDict(threading.local, MutableMapping):
-    def __init__(self):
-        super().__init__()
-        self.values = {}
-
-    def __getitem__(self, k):
-        return self.values[k]
-
-    def __setitem__(self, k, v):
-        self.values[k] = v
-
-    def __delitem__(self, k):
-        del self.values[k]
-
-    def __iter__(self):
-        return iter(self.values)
-
-    def __len__(self):
-        return len(self.values)
-
-
 _appmap_tls = ContextVar("tls")
 
 
@@ -73,24 +57,57 @@ def appmap_tls():
         return _appmap_tls.get()
 
 
+@contextmanager
+def appmap_tls_context():
+    token = _appmap_tls.set({})
+    try:
+        yield
+    finally:
+        _appmap_tls.reset(token)
+
+
 def fqname(cls):
     return "%s.%s" % (cls.__module__, cls.__qualname__)
 
-
-def split_function_name(fn):
+class FqFnName:
     """
-    Given a method, return a tuple containing its fully-qualified
-    class name and the method name.
+    FqFnName makes it easy to reference the parts of the fully-qualified name of a callable.
     """
-    qualname = fn.__qualname__
-    if "." in qualname:
-        class_name, fn_name = qualname.rsplit(".", 1)
-        class_name = "%s.%s" % (fn.__module__, class_name)
-    else:
-        class_name = fn.__module__
-        fn_name = qualname
-    return (class_name, fn_name)
 
+    def __init__(self, fn: Callable[..., Any]):
+
+        self._modname = fn.__module__
+        qualname = fn.__qualname__
+        if "." in qualname:
+            self._scope = Scope.CLASS
+            self._class_name, self._fn_name = qualname.rsplit(".", 1)
+        else:
+            self._scope = Scope.MODULE
+            self._class_name = None
+            self._fn_name = qualname
+
+    @property
+    def scope(self) -> Scope:
+        """The scope of the Callable, i.e. whether it's contained in a module or a class."""
+        return self._scope
+
+    @property
+    def fqmod(self):
+        return self._modname
+
+    @property
+    def fqclass(self):
+        return self._modname if self._class_name is None else f"{self._modname}.{self._class_name}"
+
+    @property
+    def fqfn(self):
+        return (self.fqclass, self._fn_name)
+
+    @property
+    def fn_name(self):
+        return self._fn_name
+
+FqFnName(fqname)
 
 def root_relative_path(path):
     """Returns the path relative to the current root_dir.
diff --git a/_appmap/web_framework.py b/_appmap/web_framework.py
index 6cb26efb..b5e67aa8 100644
--- a/_appmap/web_framework.py
+++ b/_appmap/web_framework.py
@@ -34,6 +34,8 @@
 # parsing the body of an application/json request:
 JSON_ERRORS = (JSONDecodeError, AttributeError, TypeError, ValueError)
 
+REQUEST_ENABLED_ATTR = "_appmap_request_enabled"
+REMOTE_ENABLED_ATTR = "_appmap_remote_enabled"
 
 class TemplateEvent(Event):  # pylint: disable=too-few-public-methods
     """A special call event that records template rendering."""
@@ -140,12 +142,13 @@ def before_request_main(self, rec, req: Any) -> Tuple[float, int]:
         raise NotImplementedError
 
     def after_request_main(self, rec, status, headers, start, call_event_id) -> None:
+
         duration = time.monotonic() - start
         return_event = HttpServerResponseEvent(
             parent_id=call_event_id,
             elapsed=duration,
             status_code=status,
-            headers=dict(headers.items()),
+            headers=headers,
         )
         rec.add_event(return_event)
 
@@ -237,7 +240,7 @@ def middleware_present(self):
 
     @abstractmethod
     def insert_middleware(self):
-        """Insert the AppMap middleware."""
+        """Insert the AppMap middleware. Optionally return a new instance of the app."""
 
     @abstractmethod
     def remote_enabled(self):
@@ -245,11 +248,13 @@ def remote_enabled(self):
 
     def run(self):
         if not self.middleware_present():
-            self.insert_middleware()
+            return self.insert_middleware()
 
         if self.remote_enabled() and not self.debug:
             self._show_warning()
 
+        return None
+
     def _show_warning(self):
         # The user has explicitly asked for remote recording to be enabled in production. Let them
         # know this probably isn't a good idea.
diff --git a/appmap/__init__.py b/appmap/__init__.py
index a2a83a47..65e38973 100644
--- a/appmap/__init__.py
+++ b/appmap/__init__.py
@@ -1,4 +1,5 @@
 """AppMap recorder for Python"""
+
 from _appmap import generation  # noqa: F401
 from _appmap.env import Env  # noqa: F401
 from _appmap.importer import instrument_module  # noqa: F401
@@ -9,13 +10,21 @@
 try:
     from . import django  # noqa: F401
 except ImportError:
-    # not using django
     pass
 
 try:
     from . import flask  # noqa: F401
 except ImportError:
-    # not using flask
+    pass
+
+try:
+    from . import fastapi  # noqa: F401
+except ImportError:
+    pass
+
+try:
+    from . import uvicorn  # noqa: F401
+except ImportError:
     pass
 
 # Note: pytest integration is configured as a pytest plugin, so it doesn't need to be imported here
diff --git a/appmap/command/appmap_agent_status.py b/appmap/command/appmap_agent_status.py
index c6edf388..3d36745a 100644
--- a/appmap/command/appmap_agent_status.py
+++ b/appmap/command/appmap_agent_status.py
@@ -1,8 +1,7 @@
 import json
 import sys
 from argparse import ArgumentParser
-
-from importlib_metadata import PackageNotFoundError, distribution, version
+from importlib.metadata import PackageNotFoundError, distribution, version
 
 from _appmap.configuration import Config
 from _appmap.env import Env
diff --git a/appmap/command/appmap_agent_validate.py b/appmap/command/appmap_agent_validate.py
index 6a538045..526d3b36 100644
--- a/appmap/command/appmap_agent_validate.py
+++ b/appmap/command/appmap_agent_validate.py
@@ -1,7 +1,7 @@
 import json
 import sys
+from importlib.metadata import PackageNotFoundError, version
 
-from importlib_metadata import PackageNotFoundError, version
 from packaging.version import parse
 
 from _appmap.py_version_check import AppMapPyVerException, check_py_version
diff --git a/appmap/django.py b/appmap/django.py
index 56b4b83c..2e393c0d 100644
--- a/appmap/django.py
+++ b/appmap/django.py
@@ -229,7 +229,7 @@ def __call__(self, request):
             request.method,
             request.build_absolute_uri(),
             response.status_code,
-            response,
+            response.headers,
             start,
             call_event_id,
         )
diff --git a/appmap/fastapi.py b/appmap/fastapi.py
new file mode 100644
index 00000000..949c7a72
--- /dev/null
+++ b/appmap/fastapi.py
@@ -0,0 +1,194 @@
+import sys
+import time
+from importlib.metadata import version
+from urllib.parse import urlunparse
+
+import fastapi
+from starlette.middleware.base import BaseHTTPMiddleware
+
+from _appmap import utils, wrapt
+from _appmap.env import Env
+from _appmap.event import HttpServerRequestEvent
+from _appmap.fastapi import app as fastapi_remote
+from _appmap.importer import Filterable, FilterableFn, Importer
+from _appmap.metadata import Metadata
+from _appmap.utils import appmap_tls_context, values_dict
+from _appmap.web_framework import (
+    JSON_ERRORS,
+    REMOTE_ENABLED_ATTR,
+    REQUEST_ENABLED_ATTR,
+    AppmapMiddleware,
+    MiddlewareInserter,
+)
+
+logger = Env.current.getLogger(__name__)
+
+
+def _add_api_route(wrapped, _, args, kwargs):
+    if not Env.current.enabled:
+        wrapped(*args, **kwargs)
+        return
+
+    fn = args[1]
+
+    fqn = utils.FqFnName(fn)
+    scope = Filterable(fqn.scope, fqn.fqclass, None)
+
+    filterable_fn = FilterableFn(scope, fn, fn)
+    logger.debug("_add_api_route, fn: %s", filterable_fn.fqname)
+    instrumented_fn = Importer.instrument_function(fqn.fn_name, filterable_fn)
+
+    if instrumented_fn != filterable_fn.obj:
+        instrumented_fn = wrapt.FunctionWrapper(fn, instrumented_fn)
+    wrapped(args[0], instrumented_fn, **kwargs)
+
+
+if Env.current.enabled:
+    wrapt.wrap_function_wrapper("fastapi.routing", "APIRouter.add_api_route", _add_api_route)
+
+
+_REQUEST_EVENT_ATTR = "_appmap_server_request_event"
+_MAX_JSON_LENGTH = 2048
+class Middleware(AppmapMiddleware, BaseHTTPMiddleware):
+
+    def __init__(self, app, remote_enabled=None):
+        super().__init__("FastAPI")
+        BaseHTTPMiddleware.__init__(self, app)
+        self._json = None
+        self._remote_enabled = remote_enabled
+
+    def init_app(self):
+        # pylint: disable=import-outside-toplevel
+        from fastapi.middleware.wsgi import WSGIMiddleware
+        from starlette.routing import Mount, Router
+
+        # pylint: enable=import-outside-toplevel
+
+        routes = [Mount("/", Middleware(self.app))]
+        setattr(self.app, REQUEST_ENABLED_ATTR, True)
+
+        if self._remote_enabled is not None:
+            enable_by_default = "true" if self._remote_enabled else "false"
+        else:
+            enable_by_default = "false"
+
+        remote_enabled = Env.current.enables("remote", enable_by_default)
+        if remote_enabled:
+            routes.insert(0, Mount("/_appmap", fastapi_remote))
+        setattr(self.app, REMOTE_ENABLED_ATTR, remote_enabled)
+
+        return Router(routes=routes)
+
+    def before_request_main(self, rec, req: fastapi.Request):
+        self.add_framework_metadata()
+        start = time.monotonic()
+        scope = req.scope
+        scope[_REQUEST_EVENT_ATTR] = call_event = HttpServerRequestEvent(
+            request_method=req.method,
+            path_info=scope["path"],
+            message_parameters={},
+            headers=req.headers,
+            protocol=f"{scope['scheme'].upper()}/{scope['http_version']}",
+        )
+        rec.add_event(call_event)
+
+        return start, call_event.id
+
+    async def dispatch(self, request, call_next):
+        with appmap_tls_context():
+            response = await self._dispatch(request, call_next)
+        return response
+
+    async def _dispatch(self, request, call_next):
+        if not self.should_record:
+            response = await call_next(request)
+            return response
+
+        await self._parse_json(request)
+
+        rec, start, call_event_id = self.before_request_hook(request)
+
+        try:
+            response = await call_next(request)
+        except:
+            self.on_exception(rec, start, call_event_id, sys.exc_info())
+            raise
+
+        self._update_request_event(request)
+
+        parsed = request.url.components
+        baseurl = urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", ""))
+        self.after_request_hook(
+            request.url.path,
+            request.method,
+            baseurl,
+            response.status_code,
+            response.headers,
+            start,
+            call_event_id,
+        )
+        return response
+
+    async def _parse_json(self, request):
+        content_length = int(request.headers.get("Content-Length", 0))
+        json_content = request.headers.get("Content-Type", "").startswith("application/json")
+        if not json_content or not 0 < content_length <= _MAX_JSON_LENGTH:
+            return
+
+        # Calling Request.json loads and caches the entire body. The cache
+        # will be used when any code subsequently tries to access the body
+        # in any way (e.g. Request.stream, Request.body, etc)
+        try:
+            self._json = await request.json()
+            if not isinstance(self._json, dict):
+                # parseable, but not a JSON object
+                self._json = None
+        except JSON_ERRORS:
+            # parsing failed, igore
+            pass
+
+    def _update_request_event(self, request):
+        # This updates the http_server_request event that was previously added
+        # to the recording. This is ok for now, because we haven't done anything
+        # with the events, e.g. streamed them to disk.
+        #
+        # If, at some point in the future, we implement some sort of
+        # checkpointing, we'll need to change this so it adds the event to the
+        # recording's `eventUpdates` instead.
+        scope = request.scope
+        if "route" not in scope:
+            return
+
+        request_event = scope[_REQUEST_EVENT_ATTR]
+        route = scope["route"]
+        request_event.normalized_path_info = route.path_format
+        query_params = {k: request.query_params.getlist(k) for k in request.query_params.keys()}
+        if self._json is not None:
+            for k, v in self._json.items():
+                if k in query_params:
+                    query_params[k].append(v)
+                else:
+                    query_params[k] = [v]
+
+        params = values_dict(query_params.items())
+        # path_params are orthogonal to query_params, so update is ok
+        params.update(request.path_params)
+        request_event.message_parameters = params
+
+    def add_framework_metadata(self):
+        Metadata.add_framework("FastAPI", version("fastapi"))
+
+
+class FastAPIInserter(MiddlewareInserter):
+    def __init__(self, app, remote_enabled):
+        super().__init__(remote_enabled)
+        self.app = app
+
+    def middleware_present(self):
+        return hasattr(self.app, REQUEST_ENABLED_ATTR)
+
+    def insert_middleware(self):
+        return Middleware(self.app, self.debug).init_app()
+
+    def remote_enabled(self):
+        return getattr(self.app, REMOTE_ENABLED_ATTR, None)
diff --git a/appmap/flask.py b/appmap/flask.py
index ef81bd06..ebc56df6 100644
--- a/appmap/flask.py
+++ b/appmap/flask.py
@@ -1,8 +1,7 @@
 import re
 import time
+from importlib.metadata import version
 
-import flask
-import flask.cli
 import jinja2
 from flask import g, got_request_exception, request, request_finished, request_started
 from flask.cli import ScriptInfo
@@ -15,7 +14,13 @@
 from _appmap.flask import app as remote_recording_app
 from _appmap.metadata import Metadata
 from _appmap.utils import patch_class, values_dict
-from _appmap.web_framework import JSON_ERRORS, AppmapMiddleware, MiddlewareInserter
+from _appmap.web_framework import (
+    JSON_ERRORS,
+    REMOTE_ENABLED_ATTR,
+    REQUEST_ENABLED_ATTR,
+    AppmapMiddleware,
+    MiddlewareInserter,
+)
 from _appmap.web_framework import TemplateHandler as BaseTemplateHandler
 
 try:
@@ -55,10 +60,6 @@ def request_params(req):
 NP_PARAMS = re.compile(r"<Rule '(.*?)'")
 NP_PARAM_DELIMS = str.maketrans("<>", "{}")
 
-_REQUEST_ENABLED_ATTR = "_appmap_request_enabled"
-_REMOTE_ENABLED_ATTR = "_appmap_remote_enabled"
-
-
 class AppmapFlask(AppmapMiddleware):
     """
     A Flask extension to add remote recording to an application.
@@ -70,33 +71,36 @@ class AppmapFlask(AppmapMiddleware):
     from appmap.flask import AppmapFlask
 
     app = new Flask(__Name__)
-    AppmapFlask().init_app(app)
+    AppmapFlask(app).init_app()
     ```
     """
 
-    def __init__(self):
+    def __init__(self, app):
         super().__init__("Flask")
+        self.app = app
 
-    def init_app(self, app):
-        enable_by_default = "true" if app.debug else "false"
+    def init_app(self):
+        enable_by_default = "true" if self.app.debug else "false"
         remote_enabled = Env.current.enables("remote", enable_by_default)
         if remote_enabled:
             logger.debug("Remote recording is enabled (Flask)")
-            app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/_appmap": remote_recording_app})
-        setattr(app, _REMOTE_ENABLED_ATTR, remote_enabled)
+            self.app.wsgi_app = DispatcherMiddleware(
+                self.app.wsgi_app, {"/_appmap": remote_recording_app}
+            )
+        setattr(self.app, REMOTE_ENABLED_ATTR, remote_enabled)
 
-        request_started.connect(self.request_started, app, weak=False)
-        request_finished.connect(self.request_finished, app, weak=False)
-        got_request_exception.connect(self.got_request_exception, app, weak=False)
+        request_started.connect(self.request_started, self.app, weak=False)
+        request_finished.connect(self.request_finished, self.app, weak=False)
+        got_request_exception.connect(self.got_request_exception, self.app, weak=False)
 
-        setattr(app, _REQUEST_ENABLED_ATTR, True)
+        setattr(self.app, REQUEST_ENABLED_ATTR, True)
 
     def request_started(self, _, **__):
         # request context is bound, we can use flask.request now:
         self.before_request_hook(request)
 
     def before_request_main(self, rec, req):
-        Metadata.add_framework("flask", flask.__version__)
+        Metadata.add_framework("flask", version("flask"))
         np = None
         if req.url_rule:
             # pragma pylint: disable=line-too-long
@@ -167,13 +171,13 @@ def __init__(self, app):
         self.app = app
 
     def middleware_present(self):
-        return hasattr(self.app, _REQUEST_ENABLED_ATTR)
+        return hasattr(self.app, REQUEST_ENABLED_ATTR)
 
     def insert_middleware(self):
-        AppmapFlask().init_app(self.app)
+        AppmapFlask(self.app).init_app()
 
     def remote_enabled(self):
-        return getattr(self.app, _REMOTE_ENABLED_ATTR, None)
+        return getattr(self.app, REMOTE_ENABLED_ATTR, None)
 
 
 def install_extension(wrapped, _, args, kwargs):
diff --git a/appmap/sqlalchemy.py b/appmap/sqlalchemy.py
index 6b49d655..3358b7c4 100644
--- a/appmap/sqlalchemy.py
+++ b/appmap/sqlalchemy.py
@@ -1,8 +1,8 @@
 """SQL statement capture for SQLAlchemy."""
 
 import time
+from importlib.metadata import version
 
-import sqlalchemy
 from sqlalchemy import event
 from sqlalchemy.engine import Engine
 
@@ -21,7 +21,7 @@ def capture_sql_call(conn, cursor, statement, parameters, context, executemany):
         # Don't record this query in the appmap.
         pass
     elif Recorder.get_enabled():
-        Metadata.add_framework("SQLAlchemy", sqlalchemy.__version__)
+        Metadata.add_framework("SQLAlchemy", version("sqlalchemy"))
         if executemany:
             # Sometimes the same query is executed with different parameter sets.
             # Instead of substituting them all, just say how many times it was run.
diff --git a/appmap/uvicorn.py b/appmap/uvicorn.py
new file mode 100644
index 00000000..edb373f9
--- /dev/null
+++ b/appmap/uvicorn.py
@@ -0,0 +1,27 @@
+# uvicorn integration
+from uvicorn.config import Config
+
+from _appmap import wrapt
+from _appmap.env import Env
+
+
+def install_extension(wrapped, config, args, kwargs):
+    wrapped(*args, **kwargs)
+    try:
+        # pylint: disable=import-outside-toplevel
+        from .fastapi import FastAPIInserter
+
+        # pylint: enable=import-outside-toplevel
+
+        app = config.loaded_app
+        if app:
+            # uvicorn doc recommends running with `--reload` in development, so use
+            # that to decide whether to enable remote recording
+            config.loaded_app = FastAPIInserter(config.loaded_app, config.reload).run()
+    except ImportError:
+        # Not FastAPI
+        pass
+
+
+if Env.current.enabled:
+    Config.load = wrapt.wrap_function_wrapper("uvicorn.config", "Config.load", install_extension)
diff --git a/pylintrc b/pylintrc
index 7f95670e..06d340d4 100644
--- a/pylintrc
+++ b/pylintrc
@@ -1,6 +1,6 @@
 [MAIN]
 # Specify a score threshold under which the program will exit with error.
-fail-under=9.82
+fail-under=9.83
 
 
 # Analyse import fallback blocks. This can be used to support both Python 2 and
diff --git a/pyproject.toml b/pyproject.toml
index af57426e..4bc3d575 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -23,7 +23,7 @@ classifiers = [
 ]
 
 include = [
-    'appmap.pth',
+    { path = 'appmap.pth', format = ['sdist','wheel'] },
     { path = '_appmap/test/**/*', format = 'sdist' }
 ]
 
@@ -41,7 +41,6 @@ packages = [
 python = "^3.8"
 PyYAML = ">=5.3.0"
 inflection = ">=0.3.0"
-importlib-metadata = ">=0.8"
 importlib-resources = "^5.4.0"
 packaging = ">=19.0"
 # If you include "Django" as an optional dependency here, you'll be able to use poetry to install it
@@ -51,34 +50,35 @@ packaging = ">=19.0"
 # So, if you'd like to run the tests outside of tox, run `pip install -r requirements-dev.txt` to
 # install it and the rest of the dev dependencies.
 
-[tool.poetry.dev-dependencies]
+[tool.poetry.group.dev.dependencies]
+SQLAlchemy = "^1.4.11"
+Twisted = "^22.4.0"
+asgiref = "^3.7.2"
+black = "^24.2.0"
+coverage = "^5.3"
+flake8 = "^3.8.4"
 httpretty = "^1.0.5"
-pytest = "^7.3.2"
-pytest-randomly = "^3.5.0"
+isort = "^5.10.1"
+pprintpp = ">=0.4.0"
+pyfakefs = "^5.3.5"
 pylint = "^2.6.0"
+pylint-exit = "^1.2.0"
+pytest = "^7.3.2"
 pytest-django = "~4.7"
-flake8 = "^3.8.4"
-pyfakefs = "^5.3.5"
-pprintpp = ">=0.4.0"
-coverage = "^5.3"
 pytest-mock = "^3.5.1"
-SQLAlchemy = "^1.4.11"
-tox = "^3.22.0"
-Twisted = "^22.4.0"
+pytest-randomly = "^3.5.0"
+pytest-shell-utilities = "^1.8.0"
+pytest-xprocess = "^0.23.0"
+python-decouple = "^3.5"
 requests = "^2.25.1"
-
+tox = "^3.22.0"
 # v2.30.0 of "requests" depends on urllib3 v2, which breaks the tests for http_client_requests. Pin
 # to v1 until this gets fixed.
 urllib3 = "^1"
-
-python-decouple = "^3.5"
-
-[tool.poetry.group.dev.dependencies]
-black = "^22.8.0"
-isort = "^5.10.1"
-pylint = "^2.16.2"
-pytest-shell-utilities = "^1.8.0"
-pylint-exit = "^1.2.0"
+uvicorn = "^0.27.1"
+fastapi = "^0.110.0"
+httpx = "^0.27.0"
+pytest-env = "^1.1.3"
 
 [build-system]
 requires = ["poetry-core>=1.1.0"]
@@ -93,7 +93,7 @@ appmap-agent-status = "appmap.command.appmap_agent_status:run"
 appmap-agent-validate = "appmap.command.appmap_agent_validate:run"
 
 [tool.black]
-line-length = 100
+line-length = 102
 extend-exclude = '''
 /(
   | vendor
diff --git a/requirements-dev.txt b/requirements-dev.txt
index cebc1a3f..c2d5913e 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -1,5 +1,7 @@
 #requirements-dev.txt
 tox
 django
-flask >=2, < 3
-pytest-django<4.8
\ No newline at end of file
+flask >=2, <= 3
+pytest-django<4.8
+fastapi
+httpx
\ No newline at end of file
diff --git a/tox.ini b/tox.ini
index 18e00615..aa889dd4 100644
--- a/tox.ini
+++ b/tox.ini
@@ -11,7 +11,6 @@ allowlist_externals =
 
 deps=
     poetry
-    pytest-django <4.8
     web: Django >=4.0, <5.0
     web: Flask >=3.0
     flask2: Flask >= 2.0, <3.0


Test patch:
diff --git a/_appmap/test/__init__.py b/_appmap/test/__init__.py
index e69de29b..c8281598 100644
--- a/_appmap/test/__init__.py
+++ b/_appmap/test/__init__.py
@@ -0,0 +1,5 @@
+import pytest
+
+# Make sure assertions in web_framework get rewritten (e.g. to show
+# diffs in generated appmaps)
+pytest.register_assert_rewrite("_appmap.test.web_framework")
diff --git a/_appmap/test/appmap_test_base.py b/_appmap/test/appmap_test_base.py
index f20afaea..161bce39 100644
--- a/_appmap/test/appmap_test_base.py
+++ b/_appmap/test/appmap_test_base.py
@@ -1,6 +1,7 @@
 import json
 import platform
 import re
+from importlib.metadata import version as dist_version
 from operator import itemgetter
 
 import pytest
@@ -60,7 +61,7 @@ def normalize_metadata(metadata):
             for f in frameworks:
                 if f["name"] == "pytest":
                     v = f.pop("version")
-                    assert v == pytest.__version__
+                    assert v == dist_version("pytest")
 
     def normalize_appmap(self, generated_appmap):
         """
diff --git a/_appmap/test/bin/server_runner b/_appmap/test/bin/server_runner
new file mode 100755
index 00000000..5fb4202d
--- /dev/null
+++ b/_appmap/test/bin/server_runner
@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+
+set -x
+
+cd "$1"; shift
+
+set -a
+PYTHONUNBUFFERED=1
+APPMAP_OUTPUT_DIR=/tmp
+PYTHONPATH=./init
+
+exec $@
diff --git a/_appmap/test/conftest.py b/_appmap/test/conftest.py
index a9f5385b..a95100dc 100644
--- a/_appmap/test/conftest.py
+++ b/_appmap/test/conftest.py
@@ -1,13 +1,21 @@
 import importlib
+import os
+import socket
 import sys
 from distutils.dir_util import copy_tree
-from functools import partialmethod
+from functools import partial, partialmethod
+from pathlib import Path
+from typing import Any
 
 import pytest
 import yaml
+from attr import dataclass
+from xprocess import ProcessStarter
 
 import _appmap
 import appmap
+from _appmap.env import Env
+from _appmap.test.web_framework import TEST_HOST, TEST_PORT
 from appmap import generation
 
 from .. import utils
@@ -128,3 +136,76 @@ def _generate(check_fn, method_name):
         return generation.dump(rec)
 
     return _generate
+
+
+@dataclass
+class ServerInfo:
+    name: str = ""
+    debug: bool = False
+    host: str = ""
+    port: int = 0
+    cmd: str = ""
+    pattern: str = ""
+    env: dict = {}
+    factory: Any = None
+
+
+class _ServerStarter(ProcessStarter):
+    @property
+    def args(self):
+        return self._args
+
+    @property
+    def pattern(self):
+        return self._pattern
+
+    def startup_check(self):
+        try:
+            s = socket.socket()
+            s.connect((self._host, self._port))
+            return True
+        except ConnectionRefusedError:
+            pass
+        return False
+
+    def __init__(self, info: ServerInfo, controldir, xprocess):
+        super().__init__(controldir, xprocess)
+        self._host = info.host
+        self._port = info.port
+        # Can't set popen_kwargs["cwd"] on a ProcessStarter until
+        # https://github.com/pytest-dev/pytest-xprocess/issues/89 is fixed.
+        #
+        # In the meantime, pass the desired directory to server_runner, which
+        # will handle changing the working directory.
+        self._args = [
+            (Path(__file__).parent / "bin" / "server_runner").as_posix(),
+            (Path(__file__).parent / "data" / info.name).as_posix(),
+            f"{Path(sys.executable).as_posix()} {info.cmd}",
+        ]
+        self._pattern = info.pattern
+        self.env = {**info.env}
+        self.terminate_on_interrupt = True
+
+
+def server_starter(info, name, cmd, pattern, env=None):
+    def _starter(controldir, xprocess):
+        info.name = name
+        info.cmd = cmd
+        if env is not None:
+            info.env = {**env, **info.env}
+        info.pattern = pattern
+        return _ServerStarter(info, controldir, xprocess)
+
+    return _starter
+
+
+@pytest.fixture(name="server_base")
+def server_base_fixture(request):
+    marker = request.node.get_closest_marker("server")
+    debug = marker.kwargs.get("debug", False)
+    server_env = os.environ.copy()
+    server_env.update(marker.kwargs.get("env", {}))
+
+    info = ServerInfo(debug=debug, host=TEST_HOST, port=TEST_PORT, env=server_env)
+    info.factory = partial(server_starter, info)
+    return info
diff --git a/_appmap/test/data/django/appmap.yml b/_appmap/test/data/django/appmap.yml
index 679a6000..4dea5e18 100644
--- a/_appmap/test/data/django/appmap.yml
+++ b/_appmap/test/data/django/appmap.yml
@@ -1,4 +1,4 @@
 name: test_app
 packages:
-- path: app
+- path: djangoapp
 
diff --git a/_appmap/test/data/django/app/__init__.py b/_appmap/test/data/django/djangoapp/__init__.py
similarity index 92%
rename from _appmap/test/data/django/app/__init__.py
rename to _appmap/test/data/django/djangoapp/__init__.py
index 37d4c6b4..a2410d96 100644
--- a/_appmap/test/data/django/app/__init__.py
+++ b/_appmap/test/data/django/djangoapp/__init__.py
@@ -6,7 +6,7 @@
 django.conf.settings.configure(
     DATABASES={"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}},
     MIDDLEWARE=["django.middleware.http.ConditionalGetMiddleware"],
-    ROOT_URLCONF="app.urls",
+    ROOT_URLCONF="djangoapp.urls",
     TEMPLATES=[
         {
             "BACKEND": "django.template.backends.django.DjangoTemplates",
diff --git a/_appmap/test/data/django/app/app.py b/_appmap/test/data/django/djangoapp/djangoapp.py
similarity index 100%
rename from _appmap/test/data/django/app/app.py
rename to _appmap/test/data/django/djangoapp/djangoapp.py
diff --git a/_appmap/test/data/django/app/hello_world.html b/_appmap/test/data/django/djangoapp/hello_world.html
similarity index 100%
rename from _appmap/test/data/django/app/hello_world.html
rename to _appmap/test/data/django/djangoapp/hello_world.html
diff --git a/_appmap/test/data/django/app/middleware.py b/_appmap/test/data/django/djangoapp/middleware.py
similarity index 100%
rename from _appmap/test/data/django/app/middleware.py
rename to _appmap/test/data/django/djangoapp/middleware.py
diff --git a/_appmap/test/data/django/app/settings.py b/_appmap/test/data/django/djangoapp/settings.py
similarity index 92%
rename from _appmap/test/data/django/app/settings.py
rename to _appmap/test/data/django/djangoapp/settings.py
index afea0c3b..63042670 100644
--- a/_appmap/test/data/django/app/settings.py
+++ b/_appmap/test/data/django/djangoapp/settings.py
@@ -6,7 +6,7 @@
 ALLOWED_HOSTS = ["*"]
 # Must set ROOT_URLCONF else we get
 # AttributeError: 'Settings' object has no attribute 'ROOT_URLCONF'
-ROOT_URLCONF = "app.urls"
+ROOT_URLCONF = "djangoapp.urls"
 
 # Turn off deprecation warning
 USE_TZ = True
diff --git a/_appmap/test/data/django/app/settings_dev.py b/_appmap/test/data/django/djangoapp/settings_dev.py
similarity index 94%
rename from _appmap/test/data/django/app/settings_dev.py
rename to _appmap/test/data/django/djangoapp/settings_dev.py
index 1637bed9..114f5c27 100644
--- a/_appmap/test/data/django/app/settings_dev.py
+++ b/_appmap/test/data/django/djangoapp/settings_dev.py
@@ -9,7 +9,7 @@
 
 # Must set ROOT_URLCONF else we get
 # AttributeError: 'Settings' object has no attribute 'ROOT_URLCONF'
-ROOT_URLCONF = "app.urls"
+ROOT_URLCONF = "djangoapp.urls"
 
 # Turn off deprecation warning
 USE_TZ = True
diff --git a/_appmap/test/data/django/app/urls.py b/_appmap/test/data/django/djangoapp/urls.py
similarity index 100%
rename from _appmap/test/data/django/app/urls.py
rename to _appmap/test/data/django/djangoapp/urls.py
diff --git a/_appmap/test/data/django/manage.py b/_appmap/test/data/django/manage.py
index 1a64b14a..c6211592 100755
--- a/_appmap/test/data/django/manage.py
+++ b/_appmap/test/data/django/manage.py
@@ -6,7 +6,7 @@
 
 def main():
     """Run administrative tasks."""
-    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "app.settings")
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "djangoapp.settings")
     try:
         from django.core.management import execute_from_command_line
     except ImportError as exc:
diff --git a/_appmap/test/data/django/pytest.ini b/_appmap/test/data/django/pytest.ini
index 98e07e24..b5755ddb 100644
--- a/_appmap/test/data/django/pytest.ini
+++ b/_appmap/test/data/django/pytest.ini
@@ -1,2 +1,2 @@
 [pytest]
-DJANGO_SETTINGS_MODULE = app.settings
+DJANGO_SETTINGS_MODULE = djangoapp.settings
diff --git a/_appmap/test/data/django/test/test_app.py b/_appmap/test/data/django/test/test_app.py
index 1d1ea0bd..8b1491ca 100644
--- a/_appmap/test/data/django/test/test_app.py
+++ b/_appmap/test/data/django/test/test_app.py
@@ -4,7 +4,7 @@
 
 @pytest.mark.parametrize(
     "mware",
-    [["app.middleware.hello_world"], ("app.middleware.hello_world",)],
+    [["djangoapp.middleware.hello_world"], ("djangoapp.middleware.hello_world",)],
 )
 def test_middleware_iterable_with_reset(client, settings, mware):
     """
diff --git a/_appmap/test/data/fastapi/appmap.yml b/_appmap/test/data/fastapi/appmap.yml
new file mode 100644
index 00000000..84e2d4da
--- /dev/null
+++ b/_appmap/test/data/fastapi/appmap.yml
@@ -0,0 +1,3 @@
+name: FastAPITest
+packages:
+- path: fastapiapp
diff --git a/_appmap/test/data/fastapi/fastapiapp/__init__.py b/_appmap/test/data/fastapi/fastapiapp/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/_appmap/test/data/fastapi/fastapiapp/main.py b/_appmap/test/data/fastapi/fastapiapp/main.py
new file mode 100644
index 00000000..95d91976
--- /dev/null
+++ b/_appmap/test/data/fastapi/fastapiapp/main.py
@@ -0,0 +1,67 @@
+"""
+Rudimentary FastAPI application for testing.
+
+NB: This should not explicitly reference the `appmap` module in any way. Doing so invalidates
+testing of record-by-default.
+"""
+# pylint: disable=missing-function-docstring
+
+from typing import List
+
+from fastapi import FastAPI, Query, Request, Response
+
+app = FastAPI()
+
+
+@app.get("/")
+def hello_world():
+    return {"Hello": "World!"}
+
+
+@app.post("/echo")
+async def echo(request: Request):
+    body = await request.body()
+    return Response(content=body, media_type="application/json")
+
+
+@app.get("/test")
+async def get_test(my_params: List[str] = Query(None)):
+    response = Response(content="testing", media_type="text/html; charset=utf-8")
+    response.headers["ETag"] = "W/01"
+    return response
+
+
+@app.post("/test")
+async def post_test(request: Request):
+    await request.json()
+    response = Response(content='{"test":true}', media_type="application/json")
+    response.headers["ETag"] = "W/01"
+    return response
+
+
+@app.get("/user/{username}")
+def get_user_profile(username):
+    # show the user profile for that user
+    return {"user": username}
+
+
+@app.get("/post/{post_id:int}")
+def get_post(post_id):
+    # show the post with the given id, the id is an integer
+    return {"post": post_id}
+
+
+@app.get("/post/{username}/{post_id:int}/summary")
+def get_user_post(username, post_id):
+    # Show the summary of a user's post
+    return {"user": username, "post": post_id}
+
+
+@app.get("/{org:int}/posts/{username}")
+def get_org_user_posts(org, username):
+    return {"org": org, "username": username}
+
+
+@app.get("/exception")
+def raise_exception():
+    raise Exception("An exception")
diff --git a/_appmap/test/data/fastapi/init/sitecustomize.py b/_appmap/test/data/fastapi/init/sitecustomize.py
new file mode 100644
index 00000000..d1fe4fec
--- /dev/null
+++ b/_appmap/test/data/fastapi/init/sitecustomize.py
@@ -0,0 +1,1 @@
+import appmap
diff --git a/_appmap/test/data/fastapi/test_app.py b/_appmap/test/data/fastapi/test_app.py
new file mode 100644
index 00000000..5355126d
--- /dev/null
+++ b/_appmap/test/data/fastapi/test_app.py
@@ -0,0 +1,14 @@
+import pytest
+from fastapi.testclient import TestClient
+from fastapiapp import app
+
+
+@pytest.fixture
+def client():
+    yield TestClient(app)
+
+
+def test_request(client):
+    response = client.get("/")
+
+    assert response.status_code == 200
diff --git a/_appmap/test/data/flask/app.py b/_appmap/test/data/flask/flaskapp.py
similarity index 100%
rename from _appmap/test/data/flask/app.py
rename to _appmap/test/data/flask/flaskapp.py
diff --git a/_appmap/test/data/flask/test_app.py b/_appmap/test/data/flask/test_app.py
index bc8db36b..eed359d1 100644
--- a/_appmap/test/data/flask/test_app.py
+++ b/_appmap/test/data/flask/test_app.py
@@ -1,11 +1,11 @@
 import pytest
-from app import app
+from flaskapp import app
 
 
-@pytest.fixture
-def client():
-    with app.test_client() as client:  # pylint: disable=no-member
-        yield client
+@pytest.fixture(name="client")
+def test_client():
+    with app.test_client() as c:  # pylint: disable=no-member
+        yield c
 
 
 def test_request(client):
diff --git a/_appmap/test/data/remote.appmap.json b/_appmap/test/data/remote.appmap.json
index 3caeef0f..15232c50 100644
--- a/_appmap/test/data/remote.appmap.json
+++ b/_appmap/test/data/remote.appmap.json
@@ -9,19 +9,15 @@
                 "protocol": "HTTP/1.1",
                 "request_method": "GET"
             },
-            "id": 1,
-            "thread_id": 1
+            "id": 1
         },
         {
             "event": "return",
             "http_server_response": {
-                "headers": { "Content-Type": "text/html; charset=utf-8" },
-                "mime_type": "text/html; charset=utf-8",
                 "status_code": 200
             },
             "id": 2,
-            "parent_id": 1,
-            "thread_id": 1
+            "parent_id": 1
         },
         {
             "event": "call",
@@ -31,19 +27,15 @@
                 "protocol": "HTTP/1.1",
                 "request_method": "GET"
             },
-            "id": 3,
-            "thread_id": 1
+            "id": 3
         },
         {
             "event": "return",
             "http_server_response": {
-                "headers": { "Content-Type": "text/html; charset=utf-8" },
-                "mime_type": "text/html; charset=utf-8",
                 "status_code": 200
             },
             "id": 4,
-            "parent_id": 3,
-            "thread_id": 1
+            "parent_id": 3
         }
     ],
     "metadata": {
diff --git a/_appmap/test/helpers.py b/_appmap/test/helpers.py
index 3ce887dc..342c35e8 100644
--- a/_appmap/test/helpers.py
+++ b/_appmap/test/helpers.py
@@ -15,3 +15,14 @@ class DictIncluding(dict):
 
     def __eq__(self, other):
         return other.items() >= self.items()
+
+
+class HeadersIncluding(dict):
+    """Like DictIncluding, but key comparison is case-insensitive."""
+
+    def __eq__(self, other):
+        for k in self.keys():
+            v = other.get(k, other.get(k.lower(), None))
+            if v is None:
+                return False
+        return True
diff --git a/_appmap/test/normalize.py b/_appmap/test/normalize.py
index fb2c5c63..77f72a71 100644
--- a/_appmap/test/normalize.py
+++ b/_appmap/test/normalize.py
@@ -1,11 +1,10 @@
 import json
 import platform
 import re
+from importlib.metadata import version as dist_version
 from operator import itemgetter
 from typing import List
 
-import pytest
-
 
 def normalize_path(path):
     """
@@ -45,16 +44,20 @@ def normalize_metadata(metadata):
         for f in frameworks:
             if f["name"] == "pytest":
                 v = f.pop("version")
-                assert v == pytest.__version__
+                assert v == dist_version("pytest")
 
 
 def normalize_headers(dct):
     """Remove some headers which are variable between implementations.
     This allows sharing tests between web frameworks, for example.
     """
-    for hdr in ["User-Agent", "Content-Length", "ETag", "Cookie", "Host"]:
-        value = dct.pop(hdr, None)
-        assert value is None or isinstance(value, str)
+    for key in list(dct.keys()):
+        value = dct.pop(key, None)
+        key = key.lower()
+        if key in ["user-agent", "content-length", "content-type", "etag", "cookie", "host"]:
+            assert value is None or isinstance(value, str)
+        else:
+            dct[key] = value
 
 
 def normalize_appmap(generated_appmap):
@@ -83,6 +86,11 @@ def normalize(dct):
             if len(dct["headers"]) == 0:
                 del dct["headers"]
         if "http_server_request" in dct:
+            # the "headers" property is optional, and will be different
+            # depending on the client sending the request. Rather than expecting
+            # particular headers, the test code should verify that other
+            # properties based on headers (e.g. "mime_type") are set correctly.
+            dct["http_server_request"].pop("headers", None)
             normalize(dct["http_server_request"])
             if "message" in dct:
                 del dct["message"]
diff --git a/_appmap/test/test_command.py b/_appmap/test/test_command.py
index 0319cec5..fbadff34 100644
--- a/_appmap/test/test_command.py
+++ b/_appmap/test/test_command.py
@@ -1,9 +1,9 @@
 import json
 import re
 from distutils.dir_util import copy_tree
+from importlib.metadata import version
 
 import pytest
-from importlib_metadata import version
 
 import _appmap
 from appmap.command import appmap_agent_init, appmap_agent_status, appmap_agent_validate
diff --git a/_appmap/test/test_configuration.py b/_appmap/test/test_configuration.py
index fca0871f..4c8eed6e 100644
--- a/_appmap/test/test_configuration.py
+++ b/_appmap/test/test_configuration.py
@@ -84,39 +84,40 @@ def test_config_no_message(caplog):
     assert caplog.text == ""
 
 
-cf = lambda: ConfigFilter(NullFilter())
+def cf():
+    return ConfigFilter(NullFilter())
 
 
 @pytest.mark.appmap_enabled(config="appmap-class.yml")
 def test_class_included():
-    f = Filterable("package1.package2.Mod1Class", None)
+    f = Filterable(None, "package1.package2.Mod1Class", None)
     assert cf().filter(f) is True
 
 
 @pytest.mark.appmap_enabled(config="appmap-func.yml")
 def test_function_included():
-    f = Filterable("package1.package2.Mod1Class.func", None)
+    f = Filterable(None, "package1.package2.Mod1Class.func", None)
     assert cf().filter(f) is True
 
 
 @pytest.mark.appmap_enabled(config="appmap-class.yml")
 def test_function_included_by_class():
-    f = Filterable("package1.package2.Mod1Class.func", None)
+    f = Filterable(None, "package1.package2.Mod1Class.func", None)
     assert cf().filter(f) is True
 
 
 @pytest.mark.appmap_enabled
 class TestConfiguration:
     def test_package_included(self):
-        f = Filterable("package1.cls", None)
+        f = Filterable(None, "package1.cls", None)
         assert cf().filter(f) is True
 
     def test_function_included_by_package(self):
-        f = Filterable("package1.package2.Mod1Class.func", None)
+        f = Filterable(None, "package1.package2.Mod1Class.func", None)
         assert cf().filter(f) is True
 
     def test_class_prefix_doesnt_match(self):
-        f = Filterable("package1_prefix.cls", None)
+        f = Filterable(None, "package1_prefix.cls", None)
         assert cf().filter(f) is False
 
 
diff --git a/_appmap/test/test_django.py b/_appmap/test/test_django.py
index 047750e9..cf22a7c4 100644
--- a/_appmap/test/test_django.py
+++ b/_appmap/test/test_django.py
@@ -3,9 +3,10 @@
 
 import json
 import os
+import socket
 import sys
 from pathlib import Path
-from threading import Thread
+from types import SimpleNamespace as NS
 
 import django
 import django.core.handlers.exception
@@ -15,28 +16,45 @@
 import pytest
 from django.template.loader import render_to_string
 from django.test.client import MULTIPART_CONTENT
+from xprocess import ProcessStarter
 
 import appmap
 import appmap.django  # noqa: F401
 from _appmap.metadata import Metadata
 
 from ..test.helpers import DictIncluding
+from .web_framework import (
+    _TestFormCapture,
+    _TestFormData,
+    _TestRecordRequests,
+    _TestRemoteRecording,
+    _TestRequestCapture,
+)
 
-# Make sure assertions in web_framework get rewritten (e.g. to show
-# diffs in generated appmaps)
-pytest.register_assert_rewrite("_appmap.test.web_framework")
+sys.path += [str(Path(__file__).parent / "data" / "django")]
 
-# pylint: disable=unused-import,wrong-import-position
-from .web_framework import TestRemoteRecording  # pyright:ignore
-from .web_framework import TestRequestCapture  # pyright: ignore
-from .web_framework import _TestRecordRequests, exec_cmd, wait_until_port_is
+# Import app just for the side-effects. It must happen after sys.path has been modified.
+import djangoapp  # pyright: ignore pylint: disable=import-error, unused-import,wrong-import-order,wrong-import-position
 
-# pylint: enable=unused-import
 
-sys.path += [str(Path(__file__).parent / "data" / "django")]
+class TestFormCapture(_TestFormCapture):
+    pass
+
+
+class TestFormTest(_TestFormData):
+    pass
+
+
+class TestRecordRequests(_TestRecordRequests):
+    pass
 
-# Import app just for the side-effects. It must happen after sys.path has been modified.
-import app  # pyright: ignore pylint: disable=import-error, unused-import,wrong-import-order,wrong-import-position
+
+class TestRemoteRecording(_TestRemoteRecording):
+    pass
+
+
+class TestRequestCapture(_TestRequestCapture):
+    pass
 
 
 @pytest.mark.django_db
@@ -68,9 +86,9 @@ def test_template(events):
     render_to_string("hello_world.html")
     assert events[0].to_dict() == DictIncluding(
         {
-            "path": "_appmap/test/data/django/app/hello_world.html",
+            "path": "_appmap/test/data/django/djangoapp/hello_world.html",
             "event": "call",
-            "defined_class": "<templates>._AppmapTestDataDjangoAppHello_WorldHtml",
+            "defined_class": "<templates>._AppmapTestDataDjangoDjangoappHello_WorldHtml",
             "method_id": "render",
             "static": False,
         }
@@ -200,55 +218,21 @@ def test_disabled(self, pytester, monkeypatch):
         assert not (pytester.path / "tmp").exists()
 
 
-class TestRecordRequestsDjango(_TestRecordRequests):
-    def server_start_thread(self, debug=True):
-        # Use appmap from our working copy, not the module installed by virtualenv. Add the init
-        # directory so the sitecustomize.py file it contains will be loaded on startup. This
-        # simulates a real installation.
-        settings = "settings_dev" if debug else "settings"
-        exec_cmd(
-            """
-export PYTHONPATH="$PWD"
-
-cd _appmap/test/data/django/
-PYTHONPATH="$PYTHONPATH:$PWD/init"
-"""
-            + f" APPMAP_OUTPUT_DIR=/tmp DJANGO_SETTINGS_MODULE=app.{settings}"
-            + " python manage.py runserver"
-            + f" 127.0.0.1:{_TestRecordRequests.server_port}"
-        )
-
-    def server_start(self, debug=True):
-        def start_with_debug():
-            self.server_start_thread(debug)
-
-        # start as background thread so running the tests can continue
-        thread = Thread(target=start_with_debug)
-        thread.start()
-        wait_until_port_is("127.0.0.1", _TestRecordRequests.server_port, "open")
-
-    def server_stop(self):
-        exec_cmd(
-            "ps -ef"
-            + "| grep -i 'manage.py runserver'"
-            + "| grep -v grep"
-            + "| awk '{ print $2 }'"
-            + "| xargs kill -9"
-        )
-        wait_until_port_is("127.0.0.1", _TestRecordRequests.server_port, "closed")
-
-    def test_record_request_appmap_enabled_requests_enabled_no_remote(self):
-        self.server_stop()  # ensure it's not running
-        self.server_start()
-        self.record_request(False)
-        self.server_stop()
-
-    def test_record_request_appmap_enabled_requests_enabled_and_remote(self):
-        self.server_stop()  # ensure it's not running
-        self.server_start()
-        self.record_request(True)
-        self.server_stop()
-
-    # it's not possible to test for
-    # appmap_not_enabled_requests_enabled_and_remote because when
-    # APPMAP=false the routes for remote recording are disabled.
+@pytest.fixture(name="server")
+def django_server(xprocess, server_base):
+    debug = server_base.debug
+    host = server_base.host
+    port = server_base.port
+    settings = "settings_dev" if debug else "settings"
+
+    name = "django"
+    pattern = f"server at http://{host}:{port}"
+    cmd = f"manage.py runserver {host}:{port}"
+    env = {"DJANGO_SETTINGS_MODULE": f"djangoapp.{settings}"}
+
+    xprocess.ensure(name, server_base.factory(name, cmd, pattern, env))
+
+    url = f"http://{server_base.host}:{port}"
+    yield NS(debug=debug, url=url)
+
+    xprocess.getinfo(name).terminate()
diff --git a/_appmap/test/test_fastapi.py b/_appmap/test/test_fastapi.py
new file mode 100644
index 00000000..c5daee90
--- /dev/null
+++ b/_appmap/test/test_fastapi.py
@@ -0,0 +1,91 @@
+import importlib
+import socket
+import sys
+from importlib.metadata import version
+from pathlib import Path
+from types import SimpleNamespace as NS
+
+import pytest
+from fastapi.testclient import TestClient
+from xprocess import ProcessStarter
+
+import appmap
+from _appmap.env import Env
+from _appmap.metadata import Metadata
+
+from .web_framework import (
+    _TestRecordRequests,
+    _TestRemoteRecording,
+    _TestRequestCapture,
+)
+
+
+# Opt in to these tests. (I couldn't find a way to DRY this up that allowed
+# pytest collection to find all these tests.)
+class TestRecordRequests(_TestRecordRequests):
+    pass
+
+
+@pytest.mark.app(remote_enabled=True)
+class TestRemoteRecording(_TestRemoteRecording):
+    def setup_method(self):
+        self.expected_thread_id = 1
+        self.expected_content_type = "application/json"
+
+
+class TestRequestCapture(_TestRequestCapture):
+    pass
+
+
+@pytest.fixture(name="app")
+def fastapi_app(data_dir, monkeypatch, request):
+    monkeypatch.syspath_prepend(data_dir / "fastapi")
+
+    Env.current.set("APPMAP_CONFIG", data_dir / "fastapi" / "appmap.yml")
+
+    from fastapiapp import main  # pyright: ignore[reportMissingImports]
+
+    importlib.reload(main)
+
+    # FastAPI doesn't provide a way to say what environment the app is running
+    # in. So, instead use a mark to indicate whether remote recording should be
+    # enabled. (When we're running as part of a server integration, we infer the
+    # environment from the server configure, e.g. "uvicorn --reload".)
+    mark = request.node.get_closest_marker("app")
+    remote_enabled = None
+    if mark is not None:
+        remote_enabled = mark.kwargs.get("remote_enabled", None)
+
+    # Add the FastAPI middleware to the app. This now happens automatically when
+    # a FastAPI app is started from uvicorn, but must be done manually
+    # otherwise.
+    return appmap.fastapi.Middleware(main.app, remote_enabled).init_app()
+
+
+@pytest.fixture(name="client")
+def fastapi_client(app):
+    yield TestClient(app, headers={})
+
+
+@pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
+def test_framework_metadata(client, events):  # pylint: disable=unused-argument
+    client.get("/")
+    assert Metadata()["frameworks"] == [{"name": "FastAPI", "version": version("fastapi")}]
+
+
+@pytest.fixture(name="server")
+def fastapi_server(xprocess, server_base):
+    debug = server_base.debug
+    host = server_base.host
+    port = server_base.port
+    reload = "--reload" if debug else ""
+
+    name = "fastapi"
+    pattern = f"Uvicorn running on http://{host}:{port}"
+    cmd = f"-m uvicorn fastapiapp.main:app {reload} --host {host} --port {port}"
+    xprocess.ensure(name, server_base.factory(name, cmd, pattern))
+
+    url = f"http://{server_base.host}:{port}"
+    yield NS(debug=debug, url=url)
+
+    xprocess.getinfo(name).terminate()
diff --git a/_appmap/test/test_flask.py b/_appmap/test/test_flask.py
index 5a324723..0d8a5a4e 100644
--- a/_appmap/test/test_flask.py
+++ b/_appmap/test/test_flask.py
@@ -3,27 +3,50 @@
 
 import importlib
 import os
-from threading import Thread
+import socket
+import sys
+from functools import partial
+from importlib.metadata import version
+from pathlib import Path
+from types import SimpleNamespace as NS
 
 import flask
 import pytest
+from attr import dataclass
+from xprocess import ProcessStarter
 
 from _appmap.env import Env
 from _appmap.metadata import Metadata
 from appmap.flask import AppmapFlask
 
 from ..test.helpers import DictIncluding
+from .web_framework import (
+    _TestFormCapture,
+    _TestFormData,
+    _TestRecordRequests,
+    _TestRemoteRecording,
+    _TestRequestCapture,
+)
 
-# Make sure assertions in web_framework get rewritten (e.g. to show
-# diffs in generated appmaps)
-pytest.register_assert_rewrite("_appmap.test.web_framework")
 
-# pylint: disable=unused-import,wrong-import-position
-from .web_framework import TestRemoteRecording  # pyright:ignore
-from .web_framework import TestRequestCapture  # pyright: ignore
-from .web_framework import _TestRecordRequests, exec_cmd, wait_until_port_is
+class TestFormCapture(_TestFormCapture):
+    pass
 
-# pylint: enable=unused-import
+
+class TestFormTest(_TestFormData):
+    pass
+
+
+class TestRecordRequests(_TestRecordRequests):
+    pass
+
+
+class TestRemoteRecording(_TestRemoteRecording):
+    pass
+
+
+class TestRequestCapture(_TestRequestCapture):
+    pass
 
 
 @pytest.fixture(name="app")
@@ -34,15 +57,15 @@ def flask_app(data_dir, monkeypatch):
 
     monkeypatch.setenv("FLASK_DEBUG", "True")
 
-    import app  # pyright: ignore pylint: disable=import-error,import-outside-toplevel
+    import flaskapp  # pyright: ignore pylint: disable=import-error,import-outside-toplevel
 
-    importlib.reload(app)
+    importlib.reload(flaskapp)
 
     # Add the AppmapFlask extension to the app. This now happens automatically when a Flask app is
     # started from the command line, but must be done manually otherwise.
-    AppmapFlask().init_app(app.app)
+    AppmapFlask(flaskapp.app).init_app()
 
-    return app.app
+    return flaskapp.app
 
 
 @pytest.fixture(name="client")
@@ -56,7 +79,7 @@ def flask_client(app):
 @pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
 def test_framework_metadata(client, events):  # pylint: disable=unused-argument
     client.get("/")
-    assert Metadata()["frameworks"] == [{"name": "flask", "version": flask.__version__}]
+    assert Metadata()["frameworks"] == [{"name": "flask", "version": version("flask")}]
 
 
 @pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
@@ -89,59 +112,25 @@ def test_template(app, events):
     )
 
 
-class TestRecordRequestsFlask(_TestRecordRequests):
-    def server_start_thread(self, debug=True):
-        # Use appmap from our working copy, not the module installed by virtualenv. Add the init
-        # directory so the sitecustomize.py file it contains will be loaded on startup. This
-        # simulates a real installation.
-        flask_debug = "FLASK_DEBUG=1" if debug else ""
-
-        exec_cmd(
-            """
-export PYTHONPATH="$PWD"
-
-cd _appmap/test/data/flask/
-PYTHONPATH="$PYTHONPATH:$PWD/init"
-"""
-            + f" APPMAP_OUTPUT_DIR=/tmp {flask_debug} FLASK_APP=app.py flask run -p "
-            + str(_TestRecordRequests.server_port)
-        )
-
-    def server_start(self, debug=True):
-        # start as background thread so running the tests can continue
-        def start_with_debug():
-            self.server_start_thread(debug)
-
-        thread = Thread(target=start_with_debug)
-        thread.start()
-        wait_until_port_is("127.0.0.1", _TestRecordRequests.server_port, "open")
-
-    def server_stop(self):
-        exec_cmd(
-            "ps -ef"
-            + "| grep -i 'flask run'"
-            + "| grep -v grep"
-            + "| awk '{ print $2 }'"
-            + "| xargs kill -9"
-        )
-        wait_until_port_is("127.0.0.1", _TestRecordRequests.server_port, "closed")
-
-    def test_record_request_appmap_enabled_requests_enabled_no_remote(self):
-        self.server_stop()  # ensure it's not running
-        self.server_start()
-        self.record_request(False)
-        self.server_stop()
+@pytest.fixture(name="server")
+def flask_server(xprocess, server_base):
+    debug = server_base.debug
+    host = server_base.host
+    port = server_base.port
 
-    def test_record_request_appmap_enabled_requests_enabled_and_remote(self):
-        self.server_stop()  # ensure it's not running
-        self.server_start()
-        self.record_request(True)
-        self.server_stop()
+    name = "flask"
+    pattern = f"Running on http://{host}:{port}"
+    cmd = f" -m flask run -p {port}"
+    env = {
+        "FLASK_APP": "flaskapp.py",
+        "FLASK_DEBUG": "1" if debug else "0",
+    }
+    xprocess.ensure(name, server_base.factory(name, cmd, pattern, env))
 
-    # it's not possible to test for
-    # appmap_not_enabled_requests_enabled_and_remote because when
-    # APPMAP=false the routes for remote recording are disabled.
+    url = f"http://{server_base.host}:{port}"
+    yield NS(debug=debug, url=url)
 
+    xprocess.getinfo(name).terminate()
 
 class TestFlaskApp:
     """
diff --git a/_appmap/test/test_recording.py b/_appmap/test/test_recording.py
index 0e53a558..1edbd153 100644
--- a/_appmap/test/test_recording.py
+++ b/_appmap/test/test_recording.py
@@ -208,7 +208,7 @@ def test_process_recording(data_dir, shell, tmp_path):
 
     appmap_dir = tmp / "tmp" / "appmap" / "process"
     appmap_files = list(appmap_dir.glob("*.appmap.json"))
-    assert len(appmap_files) == 1
+    assert len(appmap_files) == 1, "this only fails when run from VS Code?"
     actual = json.loads(appmap_files[0].read_text())
     assert len(actual["events"]) > 0
     assert len(actual["classMap"]) > 0
diff --git a/_appmap/test/test_sqlalchemy.py b/_appmap/test/test_sqlalchemy.py
index 0fe09a8a..7258f9c2 100644
--- a/_appmap/test/test_sqlalchemy.py
+++ b/_appmap/test/test_sqlalchemy.py
@@ -1,7 +1,8 @@
 """Tests for the SQLAlchemy integration."""
 
+from importlib.metadata import version
+
 import pytest
-import sqlalchemy
 from sqlalchemy import (
     Column,
     ForeignKey,
@@ -27,9 +28,7 @@ def test_sql_capture(connection, events):
             {"sql": "SELECT 1", "database_type": "sqlite"}
         )
         assert events[0].sql_query["server_version"].startswith("3.")
-        assert Metadata()["frameworks"] == [
-            {"name": "SQLAlchemy", "version": sqlalchemy.__version__}
-        ]
+        assert Metadata()["frameworks"] == [{"name": "SQLAlchemy", "version": version("sqlalchemy")}]
 
     @staticmethod
     # pylint: disable=unused-argument
diff --git a/_appmap/test/web_framework.py b/_appmap/test/web_framework.py
index 433f9410..7e7185ba 100644
--- a/_appmap/test/web_framework.py
+++ b/_appmap/test/web_framework.py
@@ -5,11 +5,8 @@
 import json
 import multiprocessing
 import os
-import socket
-import subprocess
 import time
 import traceback
-from abc import abstractmethod
 from os.path import exists
 from random import SystemRandom
 
@@ -19,10 +16,13 @@
 import appmap
 from _appmap.env import Env
 
-from ..test.helpers import DictIncluding
+from ..test.helpers import DictIncluding, HeadersIncluding
 from .normalize import normalize_appmap
 
-SR = SystemRandom()
+TEST_HOST = "127.0.0.1"
+TEST_PORT = 8000
+
+_SR = SystemRandom()
 
 
 def content_type(res):
@@ -34,7 +34,31 @@ def content_type(res):
 
 
 @pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
-class TestRequestCapture:
+class _TestFormData:
+    @staticmethod
+    @pytest.mark.parametrize("bad_json", ["bad json", 1, False, None, [2, 3]])
+    def test_post_bad_json(events, client, bad_json):
+        client.post(
+            "/test?my_param=example",
+            data=str(bad_json),
+            content_type="application/json",
+        )
+
+        assert events[0].message == [
+            DictIncluding({"name": "my_param", "class": "builtins.str", "value": "'example'"})
+        ]
+
+    @staticmethod
+    def test_post_multipart(events, client):
+        client.post("/test", data={"my_param": "example"}, content_type="multipart/form-data")
+
+        assert events[0].message == [
+            DictIncluding({"name": "my_param", "class": "builtins.str", "value": "'example'"})
+        ]
+
+
+@pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
+class _TestRequestCapture:
     """Common tests for HTTP server request and response capture."""
 
     @staticmethod
@@ -51,7 +75,7 @@ def test_http_capture(client, events):
             {"status_code": 200, "mime_type": "text/html; charset=utf-8"}
         )
 
-        assert "ETag" in response["headers"]
+        assert "etag" in map(str.lower, response["headers"].keys())
 
     @staticmethod
     def test_http_capture_post(client, events):
@@ -76,7 +100,7 @@ def test_http_capture_post(client, events):
             }
         )
 
-        assert events[0].http_server_request["headers"] == DictIncluding(
+        assert events[0].http_server_request["headers"] == HeadersIncluding(
             {"Accept": "application/json", "Accept-Language": "pl"}
         )
 
@@ -84,12 +108,12 @@ def test_http_capture_post(client, events):
     def test_post(events, client):
         client.post(
             "/test",
-            data=json.dumps({"my_param": "example"}),
-            content_type="application/json; charset=UTF-8",
+            json={"my_param": "example"},
             headers={
                 "Authorization": 'token "test-token"',
                 "Accept": "application/json",
                 "Accept-Language": "pl",
+                "Content-Type": "application/json; charset=UTF-8",
             },
         )
 
@@ -108,7 +132,7 @@ def test_post(events, client):
             }
         )
 
-        assert events[0].http_server_request["headers"] == DictIncluding(
+        assert events[0].http_server_request["headers"] == HeadersIncluding(
             {"Accept": "application/json", "Accept-Language": "pl"}
         )
 
@@ -136,20 +160,6 @@ def test_get_arr(events, client):
             ),
         ]
 
-    @staticmethod
-    def test_post_form_urlencoded(events, client):
-        client.post(
-            "/test",
-            data="my_param=example",
-            content_type="application/x-www-form-urlencoded",
-        )
-
-        assert events[0].message == [
-            DictIncluding(
-                {"name": "my_param", "class": "builtins.str", "value": "'example'"}
-            )
-        ]
-
     @staticmethod
     def test_put(events, client):
         client.put("/test", json={"my_param": "example"})
@@ -160,42 +170,16 @@ def test_put(events, client):
             )
         ]
 
-    @pytest.mark.parametrize("bad_json", ["bad json", 1, False, None, [2, 3]])
-    def test_post_bad_json(self, events, client, bad_json):
-        client.post(
-            "/test?my_param=example",
-            data=str(bad_json),
-            content_type="application/json",
-        )
-
-        assert events[0].message == [
-            DictIncluding(
-                {"name": "my_param", "class": "builtins.str", "value": "'example'"}
-            )
-        ]
-
-    @staticmethod
-    def test_post_multipart(events, client):
-        client.post(
-            "/test", data={"my_param": "example"}, content_type="multipart/form-data"
-        )
-
-        assert events[0].message == [
-            DictIncluding(
-                {"name": "my_param", "class": "builtins.str", "value": "'example'"}
-            )
-        ]
-
     @staticmethod
     def test_post_with_query(events, client):
-        client.post("/test?my_param=get", data={"my_param": "example"})
+        client.post("/test?my_param=get&my_param=an", json={"my_param": "example"})
 
         assert events[0].message == [
             DictIncluding(
                 {
                     "name": "my_param",
                     "class": "builtins.list",
-                    "value": "['get', 'example']",
+                    "value": "['get', 'an', 'example']",
                 }
             )
         ]
@@ -228,9 +212,43 @@ def test_message_path_segments(events, client):
 
 
 @pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
-class TestRemoteRecording:
+class _TestFormCapture:
+    @staticmethod
+    def test_post_form_urlencoded(events, client):
+        client.post(
+            "/test",
+            data="my_param=example",
+            content_type="application/x-www-form-urlencoded",
+        )
+
+        assert events[0].message == [
+            DictIncluding({"name": "my_param", "class": "builtins.str", "value": "'example'"})
+        ]
+
+    @staticmethod
+    def test_post_multipart(events, client):
+        client.post("/test", data={"my_param": "example"}, content_type="multipart/form-data")
+
+        assert events[0].message == [
+            DictIncluding({"name": "my_param", "class": "builtins.str", "value": "'example'"})
+        ]
+
+
+@pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REQUESTS": "false"})
+class _TestRemoteRecording:
     """Common tests for remote recording."""
 
+    @property
+    def expected_content_type(self):
+        return self._expected_content_type
+
+    @expected_content_type.setter
+    def expected_content_type(self, value):
+        self._expected_content_type = value
+
+    def setup_method(self):
+        self.expected_content_type = "text/html; charset=utf-8"
+
     @staticmethod
     # since APPMAP records by default, disable it explicitly
     @pytest.mark.appmap_enabled(appmap_enabled=False)
@@ -265,16 +283,17 @@ def test_can_only_enable_once(client):
         res = client.post("/_appmap/record")
         assert res.status_code == 409
 
-    @staticmethod
-    def test_can_record(data_dir, client):
+    def test_can_record(self, data_dir, client):
         res = client.post("/_appmap/record")
         assert res.status_code == 200
 
         res = client.get("/")
         assert res.status_code == 200
+        assert res.headers["content-type"] == self.expected_content_type
 
         res = client.get("/user/test_user")
         assert res.status_code == 200
+        assert res.headers["content-type"] == self.expected_content_type
 
         res = client.delete("/_appmap/record")
         assert res.status_code == 200
@@ -282,92 +301,49 @@ def test_can_record(data_dir, client):
         data = res.data if hasattr(res, "data") else res.content
         generated_appmap = normalize_appmap(data)
 
+        for evt in generated_appmap["events"]:
+            # Strip out thread id. The values for these vary by framework, and
+            # may not even be the same within an AppMap (e.g. FastAPI). They
+            # should always be ints, though
+            if "thread_id" in evt:
+                value = evt.pop("thread_id")
+                assert isinstance(value, int)
+
+            # Check mime_type. These also vary by framework, but will be
+            # consistent within an AppMap.
+            if "http_server_response" in evt:
+                actual_content_type = evt["http_server_response"].pop("mime_type")
+                assert actual_content_type == self.expected_content_type
+
         expected_path = data_dir / "remote.appmap.json"
         with open(expected_path, encoding="utf-8") as expected:
             expected_appmap = json.load(expected)
 
-        assert (
-            generated_appmap == expected_appmap
-        ), f"expected appmap path {expected_path}"
-
+        assert generated_appmap == expected_appmap, f"expected appmap path {expected_path}"
         res = client.delete("/_appmap/record")
         assert res.status_code == 404
 
-
-def exec_cmd(command):
-    p = subprocess.Popen(
-        command,
-        shell=True,
-        stdout=subprocess.PIPE,
-        stdin=subprocess.PIPE,
-        stderr=subprocess.STDOUT,
-    )
-    output, _ = p.communicate()
-    return output.decode()
-
-
-def port_state(address, port):
-    ret = None
-    s = socket.socket()
-    try:
-        s.connect((address, port))
-        ret = "open"
-    except Exception:  # pylint: disable=broad-except
-        ret = "closed"
-        s.close()
-    return ret
-
-
-def wait_until_port_is(address, port, desired_state):
-    max_wait_seconds = 10
-    sleep_time = 0.1
-    max_count = 1 / sleep_time * max_wait_seconds
-    count = 0
-    # don't "while True" to not lock-up the testsuite if something goes wrong
-    while count < max_count:
-        current_state = port_state(address, port)
-        if current_state == desired_state:
-            break
-
-        time.sleep(sleep_time)
-        count += 1
-
-
 class _TestRecordRequests:
     """Common tests for per-requests recording (record requests.)"""
 
-    server_port = 8000
-
-    @abstractmethod
-    def server_start(self, debug=False):
-        """Start the webserver in the background. Don't block execution."""
+    @classmethod
+    def server_url(cls):
+        return f"http://{TEST_HOST}:{TEST_PORT}"
 
-    @abstractmethod
-    def server_stop(self):
-        """Stop the webserver."""
-
-    @staticmethod
-    def server_url():
-        return "http://127.0.0.1:" + str(_TestRecordRequests.server_port)
-
-    @staticmethod
-    def record_request_thread():
+    @classmethod
+    def record_request_thread(cls):
         # I've seen occasional test failures, seemingly because the test servers can't handle the
         # barrage of requests. A tiny bit of delay still causes many, many concurrent requests, but
         # eliminates the failures.
-        time.sleep(SR.uniform(0, 0.1))
-        return requests.get(_TestRecordRequests.server_url() + "/test", timeout=30)
+        time.sleep(_SR.uniform(0, 0.1))
+        return requests.get(cls.server_url() + "/test", timeout=30)
 
-    @staticmethod
-    @pytest.mark.appmap_enabled
-    def record_request(record_remote):
+    def record_requests(self, record_remote):
         if record_remote:
             # when remote recording is enabled, this test also
             # verifies the global recorder doesn't save duplicate
             # events when per-request recording is enabled
-            response = requests.post(
-                _TestRecordRequests.server_url() + "/_appmap/record", timeout=30
-            )
+            response = requests.post(self.server_url() + "/_appmap/record", timeout=30)
             assert response.status_code == 200
 
         with concurrent.futures.ThreadPoolExecutor(
@@ -377,7 +353,7 @@ def record_request(record_remote):
             max_number_of_threads = 400
             future_to_request_number = {}
             for n in range(max_number_of_threads):
-                future = executor.submit(_TestRecordRequests.record_request_thread)
+                future = executor.submit(self.record_request_thread)
                 future_to_request_number[future] = n
 
             # wait for all threads to complete
@@ -396,11 +372,11 @@ def record_request(record_remote):
                     assert response.get_data() == b"testing"
 
                 if hasattr(response, "headers"):
-                    assert response.headers["AppMap-File-Name"]
+                    assert "AppMap-File-Name" in response.headers
                     appmap_file_name = response.headers["AppMap-File-Name"]
                 else:
                     # response.headers doesn't exist in Django 2.2
-                    assert response["AppMap-File-Name"]
+                    assert "AppMap-File-Name" in response
                     appmap_file_name = response["AppMap-File-Name"]
                 assert exists(appmap_file_name)
                 appmap_file_name_basename = appmap_file_name.split("/")[-1]
@@ -423,31 +399,34 @@ def record_request(record_remote):
 
                     # AppMap should contain only one request and response
                     http_server_requests = [
-                        event["http_server_request"]
-                        for event in recording["events"]
-                        if "http_server_request" in event
+                        event for event in recording["events"] if "http_server_request" in event
                     ]
                     http_server_responses = [
-                        event["http_server_response"]
-                        for event in recording["events"]
-                        if "http_server_response" in event
+                        event for event in recording["events"] if "http_server_response" in event
                     ]
                     assert len(http_server_requests) == 1
                     assert len(http_server_responses) == 1
+                    assert http_server_responses[0]["parent_id"] == http_server_requests[0]["id"]
 
                 os.remove(appmap_file_name)
 
-    def test_remote_disabled_in_prod(self):
-        self.server_stop()
-        self.server_start(debug=False)
-        response = requests.get(self.server_url() + "/_appmap/record", timeout=30)
+    @pytest.mark.appmap_enabled
+    @pytest.mark.server(debug=True)
+    def test_record_requests_with_remote(self, server):
+        self.record_requests(server.debug)
+
+    @pytest.mark.appmap_enabled
+    @pytest.mark.server(debug=False)
+    def test_record_requests_without_remote(self, server):
+        self.record_requests(server.debug)
+
+    @pytest.mark.server(debug=False)
+    def test_remote_disabled_in_prod(self, server):
+        response = requests.get(server.url + "/_appmap/record", timeout=30)
         assert response.status_code == 404
-        self.server_stop()
 
-    def test_remote_enabled_in_prod(self, monkeypatch):
-        self.server_stop()
-        monkeypatch.setenv("APPMAP_RECORD_REMOTE", "true")
-        self.server_start(debug=False)
-        response = requests.get(self.server_url() + "/_appmap/record", timeout=30)
+    @pytest.mark.server(debug=False, env={"APPMAP_RECORD_REMOTE": "true"})
+    @pytest.mark.appmap_enabled(env={"APPMAP_RECORD_REMOTE": "true"})
+    def test_remote_enabled_in_prod(self, server):
+        response = requests.get(server.url + "/_appmap/record", timeout=30)
         assert response.status_code == 200
-        self.server_stop()
diff --git a/appmap/pytest.py b/appmap/pytest.py
index bb4c0f21..6a1d3f13 100644
--- a/appmap/pytest.py
+++ b/appmap/pytest.py
@@ -1,3 +1,5 @@
+from importlib.metadata import version
+
 import pytest
 
 from _appmap import noappmap, testing_framework, wrapt
@@ -26,7 +28,7 @@ def __call__(self, wrapped, _, args, kwargs):
     @pytest.hookimpl
     def pytest_sessionstart(session):
         session.appmap = testing_framework.session(
-            name="pytest", recorder_type="tests", version=pytest.__version__
+            name="pytest", recorder_type="tests", version=version("pytest")
         )
 
     @pytest.hookimpl
diff --git a/pytest.ini b/pytest.ini
index 7706d12c..7d9408ce 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -1,10 +1,12 @@
 [pytest]
 markers =
+    app
     datafiles: load datafiles
     appmap_enabled
     appmap_record_requests
     example_dir
-    
+    server: arguments for server fixture
+
 testpaths = _appmap/test
 pytester_example_dir = _appmap/test/data
 
@@ -15,3 +17,6 @@ addopts = --runpytest subprocess --ignore vendor
 # We're stuck at pytest ~6.1.2. This warning got removed in a later
 # version.
 filterwarnings = ignore:testdir.copy_example is an experimental api that may change over time
+
+env =
+    APPMAP_DISABLE_LOG_FILE = true
\ No newline at end of file