diff --git a/changelog.d/19828.misc b/changelog.d/19828.misc
new file mode 100644
index 0000000000..1ed01eba2e
--- /dev/null
+++ b/changelog.d/19828.misc
@@ -0,0 +1 @@
+Port `Requester` class to Rust.
diff --git a/rust/src/lib.rs b/rust/src/lib.rs
index 8ed4e24b81..bf96422cbb 100644
--- a/rust/src/lib.rs
+++ b/rust/src/lib.rs
@@ -19,6 +19,7 @@ pub mod push;
 pub mod rendezvous;
 pub mod room_versions;
 pub mod segmenter;
+pub mod types;
 
 lazy_static! {
     static ref LOGGING_HANDLE: ResetHandle = pyo3_log::init();
@@ -71,6 +72,7 @@ fn synapse_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     msc4388_rendezvous::register_module(py, m)?;
     segmenter::register_module(py, m)?;
     room_versions::register_module(py, m)?;
+    types::register_module(py, m)?;
 
     Ok(())
 }
diff --git a/rust/src/types/mod.rs b/rust/src/types/mod.rs
new file mode 100644
index 0000000000..ffb19a83a2
--- /dev/null
+++ b/rust/src/types/mod.rs
@@ -0,0 +1,340 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2026 Element Creations Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * See the GNU Affero General Public License for more details:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ *
+ */
+
+//! Rust implementations of types from `synapse.types`.
+
+use std::collections::HashSet;
+
+use once_cell::sync::OnceCell;
+use pyo3::prelude::*;
+use pyo3::{
+    exceptions::PyKeyError,
+    types::{PyDict, PyDictMethods, PyList},
+};
+
+/// A reference to the `synapse.types.UserID` class.
+static USER_ID_CLASS: OnceCell<Py<PyAny>> = OnceCell::new();
+
+/// Access to the `synapse.types.UserID` class.
+fn user_id_class(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> {
+    Ok(USER_ID_CLASS
+        .get_or_try_init(|| -> PyResult<_> {
+            Ok(py.import("synapse.types")?.getattr("UserID")?.unbind())
+        })?
+        .bind(py))
+}
+
+/// Represents the user making a request.
+#[pyclass(frozen, skip_from_py_object, get_all, eq)]
+#[derive(Debug, PartialEq, Eq)]
+pub struct Requester {
+    /// The ID of the user making the request, in string form (see
+    /// [`Self::user`] for accessing the parsed `UserID`).
+    user_id: String,
+    /// The ID of the access token used for this request, or None for
+    /// appservices, guests, and tokens generated by the admin API
+    access_token_id: Option<i64>,
+    /// True if the user making this request is a guest
+    is_guest: bool,
+    /// Any scopes associated with the access token used for this request, or an
+    /// empty set if no token or a non-oauth token was used
+    scope: HashSet<String>,
+    /// True if the user making this request is shadow banned
+    shadow_banned: bool,
+    /// The device_id which was set at authentication time, or None for
+    /// appservices, guests, and tokens generated by the admin API
+    device_id: Option<String>,
+    /// The ID of the AS requesting on behalf of the user, or None.
+    app_service_id: Option<String>,
+    /// The entity that authenticated when making the request.
+    ///
+    /// This is different to the `user_id` when an admin user or the server is
+    /// "puppeting" the user.
+    authenticated_entity: String,
+}
+
+#[pymethods]
+impl Requester {
+    #[new]
+    #[pyo3(signature = (
+        user,
+        access_token_id,
+        is_guest,
+        scope,
+        shadow_banned,
+        device_id,
+        app_service_id,
+        authenticated_entity,
+    ))]
+    #[allow(clippy::too_many_arguments)]
+    fn new(
+        user: &Bound<'_, PyAny>,
+        access_token_id: Option<i64>,
+        is_guest: &Bound<'_, PyAny>,
+        scope: HashSet<String>,
+        shadow_banned: &Bound<'_, PyAny>,
+        device_id: Option<String>,
+        app_service_id: Option<String>,
+        authenticated_entity: String,
+    ) -> PyResult<Self> {
+        // The `user` argument should be a `UserID`, which has a `to_string` for
+        // getting the string form.
+        let user_id = user.call_method0("to_string")?.extract::<String>()?;
+
+        // The `is_guest` and `shadow_banned` arguments are expected to be
+        // Python bools, but unfortunately Synapse often passes them as truthy
+        // values (mainly due to reading from SQLite, which returns 0/1 for
+        // bools).
+        let is_guest = is_guest.is_truthy()?;
+        let shadow_banned = shadow_banned.is_truthy()?;
+
+        Ok(Requester {
+            user_id,
+            access_token_id,
+            is_guest,
+            scope,
+            shadow_banned,
+            device_id,
+            app_service_id,
+            authenticated_entity,
+        })
+    }
+
+    /// The user making the request, as a Python `UserID`.
+    #[getter]
+    fn user<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        user_id_class(py)?.call_method1("from_string", (&self.user_id,))
+    }
+
+    /// Converts self to a type that can be serialized as JSON, and then
+    /// deserialized by [`Self::deserialize`]
+    fn serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDict>> {
+        let dict = PyDict::new(py);
+        dict.set_item("user_id", &self.user_id)?;
+        dict.set_item("access_token_id", self.access_token_id)?;
+        dict.set_item("is_guest", self.is_guest)?;
+        dict.set_item("scope", PyList::new(py, &self.scope)?)?;
+        dict.set_item("shadow_banned", self.shadow_banned)?;
+        dict.set_item("device_id", self.device_id.as_deref())?;
+        // NB: the wire key is "app_server_id" (server, not service). Changing
+        // this is non-trivial as it would break replication during a rolling
+        // upgrade.
+        dict.set_item("app_server_id", self.app_service_id.as_deref())?;
+        dict.set_item("authenticated_entity", &self.authenticated_entity)?;
+        Ok(dict)
+    }
+
+    /// Converts a dict that was produced by [`Self::serialize`] back into a
+    /// [`Requester`].
+    #[staticmethod]
+    fn deserialize(py: Python<'_>, input: &Bound<'_, PyAny>) -> PyResult<Self> {
+        let user_id = input.get_item("user_id")?.extract::<String>()?;
+        let access_token_id = input
+            .get_item("access_token_id")?
+            .extract::<Option<i64>>()?;
+        let is_guest = input.get_item("is_guest")?.is_truthy()?;
+
+        // `serialize` stores the scope as a list, so extract it as a `Vec`
+        // (which accepts any sequence) and collect into a set. For backwards
+        // compatibility, "scope" is optional and defaults to an empty set if
+        // not present.
+        let scope = match input.get_item("scope") {
+            Ok(scope) => scope.extract::<Vec<String>>()?.into_iter().collect(),
+            Err(err) if err.is_instance_of::<PyKeyError>(py) => HashSet::new(),
+            Err(err) => return Err(err),
+        };
+
+        let shadow_banned = input.get_item("shadow_banned")?.is_truthy()?;
+        let device_id = input.get_item("device_id")?.extract::<Option<String>>()?;
+
+        // The wire key is "app_server_id", not "app_service_id" — see `serialize`.
+        let app_service_id = input
+            .get_item("app_server_id")?
+            .extract::<Option<String>>()?;
+
+        let authenticated_entity = input
+            .get_item("authenticated_entity")?
+            .extract::<String>()?;
+
+        Ok(Requester {
+            user_id,
+            access_token_id,
+            is_guest,
+            scope,
+            shadow_banned,
+            device_id,
+            app_service_id,
+            authenticated_entity,
+        })
+    }
+
+    fn __repr__(&self) -> String {
+        format!(
+            "Requester(user_id={}, access_token_id={:?}, is_guest={}, scope={:?}, \
+             shadow_banned={}, device_id={:?}, app_service_id={:?}, authenticated_entity={})",
+            self.user_id,
+            self.access_token_id,
+            self.is_guest,
+            self.scope,
+            self.shadow_banned,
+            self.device_id,
+            self.app_service_id,
+            self.authenticated_entity,
+        )
+    }
+}
+
+/// Called when registering modules with python.
+pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
+    let child_module = PyModule::new(py, "types")?;
+    child_module.add_class::<Requester>()?;
+
+    m.add_submodule(&child_module)?;
+
+    // We need to manually add the module to sys.modules to make `from
+    // synapse.synapse_rust.types import Requester` work.
+    py.import("sys")?
+        .getattr("modules")?
+        .set_item("synapse.synapse_rust.types", child_module)?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A `Requester` with every field populated, for use as a test fixture.
+    fn sample_requester() -> Requester {
+        Requester {
+            user_id: "@alice:example.com".to_string(),
+            access_token_id: Some(42),
+            is_guest: false,
+            scope: HashSet::from(["urn:matrix:client:api:*".to_string()]),
+            shadow_banned: false,
+            device_id: Some("ABCDEFG".to_string()),
+            app_service_id: None,
+            authenticated_entity: "@alice:example.com".to_string(),
+        }
+    }
+
+    #[test]
+    fn test_serialize() {
+        Python::initialize();
+        Python::attach(|py| -> Result<(), PyErr> {
+            let requester = sample_requester();
+            let dict = requester.serialize(py)?;
+            let dict = dict.as_any();
+
+            assert_eq!(
+                dict.get_item("user_id")?.extract::<String>()?,
+                "@alice:example.com"
+            );
+            assert_eq!(dict.get_item("access_token_id")?.extract::<i64>()?, 42);
+            assert!(!dict.get_item("is_guest")?.extract::<bool>()?);
+            assert_eq!(
+                dict.get_item("scope")?.extract::<Vec<String>>()?,
+                vec!["urn:matrix:client:api:*".to_string()]
+            );
+            assert!(!dict.get_item("shadow_banned")?.extract::<bool>()?);
+            assert_eq!(
+                dict.get_item("device_id")?.extract::<Option<String>>()?,
+                Some("ABCDEFG".to_string())
+            );
+            assert_eq!(
+                dict.get_item("authenticated_entity")?.extract::<String>()?,
+                "@alice:example.com"
+            );
+
+            // The `app_service_id` field is serialized under the wire key
+            // "app_server_id" (server, not service), and there must be no
+            // "app_service_id" key.
+            assert!(dict.get_item("app_server_id")?.is_none());
+            assert!(!dict.contains("app_service_id")?);
+
+            Ok(())
+        })
+        .unwrap();
+    }
+
+    #[test]
+    fn test_serialize_deserialize_round_trip() {
+        Python::initialize();
+        Python::attach(|py| {
+            // Use a requester that exercises the optional fields and the
+            // `app_service_id` -> "app_server_id" wire-key mapping.
+            let requester = Requester {
+                user_id: "@bob:example.com".to_string(),
+                access_token_id: None,
+                is_guest: true,
+                scope: HashSet::from(["a".to_string(), "b".to_string()]),
+                shadow_banned: true,
+                device_id: None,
+                app_service_id: Some("my_appservice".to_string()),
+                authenticated_entity: "@admin:example.com".to_string(),
+            };
+
+            let dict = requester.serialize(py).unwrap();
+            let deserialized = Requester::deserialize(py, dict.as_any()).unwrap();
+            let deserialized = Bound::new(py, deserialized).unwrap();
+
+            assert_eq!(&requester, deserialized.get());
+        });
+    }
+
+    #[test]
+    fn test_deserialize_defaults_scope_when_missing() {
+        Python::initialize();
+        Python::attach(|py| {
+            // An older serialized form may omit "scope"; it should default to
+            // an empty set rather than erroring.
+            let dict = PyDict::new(py);
+            dict.set_item("user_id", "@alice:example.com").unwrap();
+            dict.set_item("access_token_id", py.None()).unwrap();
+            dict.set_item("is_guest", false).unwrap();
+            dict.set_item("shadow_banned", false).unwrap();
+            dict.set_item("device_id", py.None()).unwrap();
+            dict.set_item("app_server_id", py.None()).unwrap();
+            dict.set_item("authenticated_entity", "@alice:example.com")
+                .unwrap();
+
+            let requester = Requester::deserialize(py, dict.as_any()).unwrap();
+            assert!(requester.scope.is_empty());
+        });
+    }
+
+    #[test]
+    fn test_deserialize_coerces_truthy_bools() {
+        Python::initialize();
+        Python::attach(|py| {
+            // SQLite returns 0/1 for booleans, so non-bool truthy values must
+            // be coerced for `is_guest` and `shadow_banned`.
+            let dict = PyDict::new(py);
+            dict.set_item("user_id", "@alice:example.com").unwrap();
+            dict.set_item("access_token_id", py.None()).unwrap();
+            dict.set_item("is_guest", 1).unwrap();
+            dict.set_item("scope", PyList::empty(py)).unwrap();
+            dict.set_item("shadow_banned", 0).unwrap();
+            dict.set_item("device_id", py.None()).unwrap();
+            dict.set_item("app_server_id", py.None()).unwrap();
+            dict.set_item("authenticated_entity", "@alice:example.com")
+                .unwrap();
+
+            let requester = Requester::deserialize(py, dict.as_any()).unwrap();
+            assert!(requester.is_guest);
+            assert!(!requester.shadow_banned);
+        });
+    }
+}
diff --git a/synapse/api/auth/base.py b/synapse/api/auth/base.py
index ff876b9d22..14e76b0cff 100644
--- a/synapse/api/auth/base.py
+++ b/synapse/api/auth/base.py
@@ -371,7 +371,9 @@ class BaseAuth:
         """
         ip_addr = request.get_client_ip_if_available()
 
-        if ip_addr and (not requester.app_service or self._track_appservice_user_ips):
+        if ip_addr and (
+            not requester.app_service_id or self._track_appservice_user_ips
+        ):
             user_agent = get_request_user_agent(request)
             access_token = self.get_access_token_from_request(request)
 
@@ -381,7 +383,7 @@ class BaseAuth:
             # table during the transition
             recorded_device_id = (
                 "dummy-device"
-                if requester.device_id is None and requester.app_service is not None
+                if requester.device_id is None and requester.app_service_id is not None
                 else requester.device_id
             )
             await self.store.insert_client_ip(
diff --git a/synapse/api/auth/internal.py b/synapse/api/auth/internal.py
index b33384c13f..28c0491140 100644
--- a/synapse/api/auth/internal.py
+++ b/synapse/api/auth/internal.py
@@ -106,8 +106,8 @@ class InternalAuth(BaseAuth):
                 parent_span.set_tag("user_id", requester.user.to_string())
                 if requester.device_id is not None:
                     parent_span.set_tag("device_id", requester.device_id)
-                if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                if requester.app_service_id is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service_id)
             return requester
 
     async def get_user_by_req_experimental_feature(
diff --git a/synapse/api/auth/mas.py b/synapse/api/auth/mas.py
index 95c6d62a9d..ed0427d6f3 100644
--- a/synapse/api/auth/mas.py
+++ b/synapse/api/auth/mas.py
@@ -311,8 +311,8 @@ class MasDelegatedAuth(BaseAuth):
                 parent_span.set_tag("user_id", requester.user.to_string())
                 if requester.device_id is not None:
                     parent_span.set_tag("device_id", requester.device_id)
-                if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                if requester.app_service_id is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service_id)
             return requester
 
     async def get_user_by_access_token(
diff --git a/synapse/api/auth/msc3861_delegated.py b/synapse/api/auth/msc3861_delegated.py
index 27ab4af805..3b37f39875 100644
--- a/synapse/api/auth/msc3861_delegated.py
+++ b/synapse/api/auth/msc3861_delegated.py
@@ -420,8 +420,8 @@ class MSC3861DelegatedAuth(BaseAuth):
                 parent_span.set_tag("user_id", requester.user.to_string())
                 if requester.device_id is not None:
                     parent_span.set_tag("device_id", requester.device_id)
-                if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                if requester.app_service_id is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service_id)
             return requester
 
     async def _wrapped_get_user_by_req(
diff --git a/synapse/api/auth_blocking.py b/synapse/api/auth_blocking.py
index 3ed47b20c4..87918e15dc 100644
--- a/synapse/api/auth_blocking.py
+++ b/synapse/api/auth_blocking.py
@@ -88,7 +88,7 @@ class AuthBlocking:
                 # We never block the server from doing actions on behalf of
                 # users.
                 return
-            if requester.app_service and not self._track_appservice_user_ips:
+            if requester.app_service_id and not self._track_appservice_user_ips:
                 # If we're authenticated as an appservice then we only block
                 # auth if `track_appservice_user_ips` is set, as that option
                 # implicitly means that application services are part of MAU
diff --git a/synapse/api/ratelimiting.py b/synapse/api/ratelimiting.py
index d6cc3d26b5..b17fcd1ef3 100644
--- a/synapse/api/ratelimiting.py
+++ b/synapse/api/ratelimiting.py
@@ -163,7 +163,12 @@ class Ratelimiter:
         if requester:
             # Disable rate limiting of users belonging to any AS that is configured
             # not to be rate limited in its registration file (rate_limited: true|false).
-            if requester.app_service and not requester.app_service.is_rate_limited():
+            app_service = (
+                self.store.get_app_service_by_id(requester.app_service_id)
+                if requester.app_service_id
+                else None
+            )
+            if app_service and not app_service.is_rate_limited():
                 return True, -1.0
 
             # Check if ratelimiting has been disabled for the user.
diff --git a/synapse/events/utils.py b/synapse/events/utils.py
index 54f662796b..8ce795052a 100644
--- a/synapse/events/utils.py
+++ b/synapse/events/utils.py
@@ -419,7 +419,7 @@ def _serialize_event(
                         and event_token_id == config.requester.access_token_id
                     )
                     or config.requester.is_guest
-                    or config.requester.app_service
+                    or config.requester.app_service_id
                 ):
                     d["unsigned"]["transaction_id"] = txn_id
 
diff --git a/synapse/handlers/admin.py b/synapse/handlers/admin.py
index 0b1251f9e3..ada799d96a 100644
--- a/synapse/handlers/admin.py
+++ b/synapse/handlers/admin.py
@@ -427,7 +427,7 @@ class AdminHandler:
 
         r = task.params.get("requester")
         assert r is not None
-        admin = Requester.deserialize(self._store, r)
+        admin = Requester.deserialize(r)
 
         user_id = task.params.get("user_id")
         assert user_id is not None
diff --git a/synapse/handlers/directory.py b/synapse/handlers/directory.py
index bdb7cba64b..91e7854f39 100644
--- a/synapse/handlers/directory.py
+++ b/synapse/handlers/directory.py
@@ -132,7 +132,11 @@ class DirectoryHandler:
                 Codes.INVALID_PARAM,
             )
 
-        service = requester.app_service
+        service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
         if service:
             if not service.is_room_alias_in_namespace(room_alias_str):
                 raise SynapseError(
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index bade219ddf..ed04547d42 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -343,7 +343,7 @@ class MessageHandler:
         Returns:
             A dict of user_id to profile info
         """
-        if not requester.app_service:
+        if not requester.app_service_id:
             # We check AS auth after fetching the room membership, as it
             # requires us to pull out all joined members anyway.
             membership, _ = await self.auth.check_user_in_room_or_world_readable(
@@ -365,12 +365,14 @@ class MessageHandler:
         # If this is an AS, double check that they are allowed to see the members.
         # This can either be because the AS user is in the room or because there
         # is a user in the room that the AS is "interested in"
-        if (
-            requester.app_service
-            and requester.user.to_string() not in users_with_profile
-        ):
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        if app_service and requester.user.to_string() not in users_with_profile:
             for uid in users_with_profile:
-                if requester.app_service.is_interested_in_user(uid):
+                if app_service.is_interested_in_user(uid):
                     break
             else:
                 # Loop fell through, AS has no interested users in room
@@ -846,7 +848,7 @@ class EventCreationHandler:
             return
 
         # exempt AS users from needing consent
-        if requester.app_service is not None:
+        if requester.app_service_id is not None:
             return
 
         user_id = requester.authenticated_entity
@@ -1425,8 +1427,10 @@ class EventCreationHandler:
             else:
                 context = await self.state.calculate_context_info(event)
 
-        if requester:
-            context.app_service = requester.app_service
+        if requester and requester.app_service_id:
+            context.app_service = self.store.get_app_service_by_id(
+                requester.app_service_id
+            )
 
         res, new_content = await self._third_party_event_rules.check_event_allowed(
             event, context
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index 8a19dba5ee..5152d0b522 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -661,8 +661,8 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
         key = (room_id,)
 
         as_id = object()
-        if requester.app_service:
-            as_id = requester.app_service.id
+        if requester.app_service_id:
+            as_id = requester.app_service_id
 
         # We first linearise by the application service (to try to limit concurrent joins
         # by application services), and then by room ID.
diff --git a/synapse/replication/http/membership.py b/synapse/replication/http/membership.py
index 8a6c971720..80dd9b7da7 100644
--- a/synapse/replication/http/membership.py
+++ b/synapse/replication/http/membership.py
@@ -89,7 +89,7 @@ class ReplicationRemoteJoinRestServlet(ReplicationEndpoint):
         remote_room_hosts = content["remote_room_hosts"]
         event_content = content["content"]
 
-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
         request.requester = requester
 
         logger.info("remote_join: %s into room: %s", user_id, room_id)
@@ -153,7 +153,7 @@ class ReplicationRemoteKnockRestServlet(ReplicationEndpoint):
         remote_room_hosts = content["remote_room_hosts"]
         event_content = content["content"]
 
-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
         request.requester = requester
 
         logger.debug("remote_knock: %s on room: %s", user_id, room_id)
@@ -219,7 +219,7 @@ class ReplicationRemoteRejectInviteRestServlet(ReplicationEndpoint):
         txn_id = content["txn_id"]
         event_content = content["content"]
 
-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
         request.requester = requester
 
         # hopefully we're now on the master, so this won't recurse!
@@ -283,7 +283,7 @@ class ReplicationRemoteRescindKnockRestServlet(ReplicationEndpoint):
         txn_id = content["txn_id"]
         event_content = content["content"]
 
-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
         request.requester = requester
 
         # hopefully we're now on the master, so this won't recurse!
diff --git a/synapse/replication/http/send_events.py b/synapse/replication/http/send_events.py
index b020a0fe7c..e0cfb0d17c 100644
--- a/synapse/replication/http/send_events.py
+++ b/synapse/replication/http/send_events.py
@@ -141,9 +141,7 @@ class ReplicationSendEventsRestServlet(ReplicationEndpoint):
                 )
                 event.internal_metadata.outlier = event_payload["outlier"]
 
-                requester = Requester.deserialize(
-                    self.store, event_payload["requester"]
-                )
+                requester = Requester.deserialize(event_payload["requester"])
                 context = EventContext.deserialize(
                     self._storage_controllers, event_payload["context"]
                 )
diff --git a/synapse/rest/client/account.py b/synapse/rest/client/account.py
index 3cb1e09f44..d1e404f0dc 100644
--- a/synapse/rest/client/account.py
+++ b/synapse/rest/client/account.py
@@ -305,7 +305,7 @@ class DeactivateAccountRestServlet(RestServlet):
 
         # allow ASes to deactivate their own users:
         # ASes don't need user-interactive auth
-        if not requester.app_service:
+        if not requester.app_service_id:
             await self.auth_handler.validate_user_via_ui_auth(
                 requester,
                 request,
diff --git a/synapse/rest/client/appservice_ping.py b/synapse/rest/client/appservice_ping.py
index 7e2ac15783..2c6ad5bcf0 100644
--- a/synapse/rest/client/appservice_ping.py
+++ b/synapse/rest/client/appservice_ping.py
@@ -55,25 +55,32 @@ class AppservicePingRestServlet(RestServlet):
         self.as_api = hs.get_application_service_api()
         self.scheduler = hs.get_application_service_scheduler()
         self.auth = hs.get_auth()
+        self.store = hs.get_datastores().main
 
     async def on_POST(
         self, request: SynapseRequest, appservice_id: str
     ) -> tuple[int, JsonDict]:
         requester = await self.auth.get_user_by_req(request)
 
-        if not requester.app_service:
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+
+        if not app_service:
             raise SynapseError(
                 HTTPStatus.FORBIDDEN,
                 "Only application services can use the /appservice/ping endpoint",
                 Codes.FORBIDDEN,
             )
-        elif requester.app_service.id != appservice_id:
+        elif app_service.id != appservice_id:
             raise SynapseError(
                 HTTPStatus.FORBIDDEN,
                 "Mismatching application service ID in path",
                 Codes.FORBIDDEN,
             )
-        elif not requester.app_service.url:
+        elif not app_service.url:
             raise SynapseError(
                 HTTPStatus.BAD_REQUEST,
                 "The application service does not have a URL set",
@@ -85,11 +92,11 @@ class AppservicePingRestServlet(RestServlet):
 
         start = time.monotonic()
         try:
-            await self.as_api.ping(requester.app_service, txn_id)
+            await self.as_api.ping(app_service, txn_id)
 
             # We got a OK response, so if the AS needs to be recovered then lets recover it now.
             # This sets off a task in the background and so is safe to execute and forget.
-            self.scheduler.txn_ctrl.force_retry(requester.app_service)
+            self.scheduler.txn_ctrl.force_retry(app_service)
         except RequestTimedOutError as e:
             raise SynapseError(
                 HTTPStatus.GATEWAY_TIMEOUT,
diff --git a/synapse/rest/client/devices.py b/synapse/rest/client/devices.py
index 4b84131d32..0231ed374d 100644
--- a/synapse/rest/client/devices.py
+++ b/synapse/rest/client/devices.py
@@ -105,7 +105,7 @@ class DeleteDevicesRestServlet(RestServlet):
             else:
                 raise e
 
-        if requester.app_service:
+        if requester.app_service_id:
             # MSC4190 can skip UIA for this endpoint
             pass
         else:
@@ -177,7 +177,7 @@ class DeviceRestServlet(RestServlet):
             else:
                 raise
 
-        if requester.app_service:
+        if requester.app_service_id:
             # MSC4190 allows appservices to delete devices through this endpoint without UIA
             # It's also allowed with MSC3861 enabled
             pass
@@ -212,7 +212,7 @@ class DeviceRestServlet(RestServlet):
         body = parse_and_validate_json_object_from_request(request, self.PutBody)
 
         # MSC4190 allows appservices to create devices through this endpoint
-        if requester.app_service:
+        if requester.app_service_id:
             created = await self.device_handler.upsert_device(
                 user_id=requester.user.to_string(),
                 device_id=device_id,
diff --git a/synapse/rest/client/directory.py b/synapse/rest/client/directory.py
index 0b334f9b0b..04941b25f3 100644
--- a/synapse/rest/client/directory.py
+++ b/synapse/rest/client/directory.py
@@ -110,14 +110,19 @@ class ClientDirectoryServer(RestServlet):
         room_alias_obj = RoomAlias.from_string(room_alias)
         requester = await self.auth.get_user_by_req(request)
 
-        if requester.app_service:
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        if app_service:
             await self.directory_handler.delete_appservice_association(
-                requester.app_service, room_alias_obj
+                app_service, room_alias_obj
             )
 
             logger.info(
                 "Application service at %s deleted alias %s",
-                requester.app_service.url,
+                app_service.url,
                 room_alias_obj.to_string(),
             )
 
@@ -199,13 +204,13 @@ class ClientAppserviceDirectoryListServer(RestServlet):
         visibility: Literal["public", "private"],
     ) -> tuple[int, JsonDict]:
         requester = await self.auth.get_user_by_req(request)
-        if not requester.app_service:
+        if not requester.app_service_id:
             raise AuthError(
                 403, "Only appservices can edit the appservice published room list"
             )
 
         await self.directory_handler.edit_published_appservice_room_list(
-            requester.app_service.id, network_id, room_id, visibility
+            requester.app_service_id, network_id, room_id, visibility
         )
 
         return 200, {}
diff --git a/synapse/rest/client/keys.py b/synapse/rest/client/keys.py
index 2c65a55ea1..463c87d92b 100644
--- a/synapse/rest/client/keys.py
+++ b/synapse/rest/client/keys.py
@@ -535,7 +535,7 @@ class SigningKeyUploadServlet(RestServlet):
         # setup, and that is allowed without UIA, per MSC3967.
         # If yes, then we need to authenticate the change.
         # MSC4190 can skip UIA for replacing cross-signing keys as well.
-        if is_cross_signing_setup and not requester.app_service:
+        if is_cross_signing_setup and not requester.app_service_id:
             # With MSC3861, UIA is not possible. Instead, the auth service has to
             # explicitly mark the master key as replaceable.
             if self.hs.config.mas.enabled:
diff --git a/synapse/rest/client/login.py b/synapse/rest/client/login.py
index a49b27d009..cfdc97b664 100644
--- a/synapse/rest/client/login.py
+++ b/synapse/rest/client/login.py
@@ -203,7 +203,11 @@ class LoginRestServlet(RestServlet):
         try:
             if login_submission["type"] == LoginRestServlet.APPSERVICE_TYPE:
                 requester = await self.auth.get_user_by_req(request)
-                appservice = requester.app_service
+                appservice = (
+                    self._main_store.get_app_service_by_id(requester.app_service_id)
+                    if requester.app_service_id
+                    else None
+                )
 
                 if appservice is None:
                     raise InvalidClientTokenError(
diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py
index 48ef42c7d6..6e00197b6d 100644
--- a/synapse/rest/client/room.py
+++ b/synapse/rest/client/room.py
@@ -336,7 +336,7 @@ class RoomStateEventRestServlet(RestServlet):
                 )
 
         origin_server_ts = None
-        if requester.app_service:
+        if requester.app_service_id:
             origin_server_ts = parse_integer(request, "ts")
 
         sticky_duration_ms: int | None = None
@@ -435,7 +435,7 @@ class RoomSendEventRestServlet(TransactionRestServlet):
         content = parse_json_object_from_request(request)
 
         origin_server_ts = None
-        if requester.app_service:
+        if requester.app_service_id:
             origin_server_ts = parse_integer(request, "ts")
 
         sticky_duration_ms: int | None = None
diff --git a/synapse/rest/client/transactions.py b/synapse/rest/client/transactions.py
index 43c7b6f993..0046b31b49 100644
--- a/synapse/rest/client/transactions.py
+++ b/synapse/rest/client/transactions.py
@@ -82,8 +82,8 @@ class HttpTransactionCache:
             assert requester.user is not None, "Guest requester must have a user ID set"
             return (path, "guest", requester.user)
 
-        elif requester.app_service is not None:
-            return (path, "appservice", requester.app_service.id)
+        elif requester.app_service_id is not None:
+            return (path, "appservice", requester.app_service_id)
 
         # Use the user ID and device ID as the transaction key.
         elif requester.device_id:
diff --git a/synapse/rest/media/create_resource.py b/synapse/rest/media/create_resource.py
index 8bef1f4c08..1b6b001b45 100644
--- a/synapse/rest/media/create_resource.py
+++ b/synapse/rest/media/create_resource.py
@@ -43,6 +43,7 @@ class CreateResource(RestServlet):
         super().__init__()
 
         self.media_repo = media_repo
+        self.store = hs.get_datastores().main
         self.clock = hs.get_clock()
         self.auth = hs.get_auth()
         self.max_pending_media_uploads = hs.config.media.max_pending_media_uploads
@@ -60,7 +61,12 @@ class CreateResource(RestServlet):
         # If the create media requests for the user are over the limit, drop them.
         await self._create_media_rate_limiter.ratelimit(requester)
 
-        if not requester.app_service or requester.app_service.is_rate_limited():
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        if not app_service or app_service.is_rate_limited():
             (
                 reached_pending_limit,
                 first_expiration_ts,
diff --git a/synapse/synapse_rust/types.pyi b/synapse/synapse_rust/types.pyi
new file mode 100644
index 0000000000..36a030fb57
--- /dev/null
+++ b/synapse/synapse_rust/types.pyi
@@ -0,0 +1,68 @@
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+from synapse.types import JsonDict, UserID
+
+class Requester:
+    def __init__(
+        self,
+        user: UserID,
+        access_token_id: int | None,
+        is_guest: bool,
+        scope: set[str],
+        shadow_banned: bool,
+        device_id: str | None,
+        app_service_id: str | None,
+        authenticated_entity: str,
+    ) -> None: ...
+    @property
+    def user_id(self) -> str:
+        """The ID of the user making the request, in string form (see `user`
+        for the parsed UserID)"""
+
+    @property
+    def user(self) -> UserID:
+        """The user making the request"""
+    @property
+    def access_token_id(self) -> int | None:
+        """The ID of the access token used for this request, or
+        None for appservices, guests, and tokens generated by the admin API"""
+    @property
+    def is_guest(self) -> bool:
+        """True if the user making this request is a guest user"""
+    @property
+    def scope(self) -> set[str]:
+        """Any scopes associated with the access token used for this request, or
+        an empty set if no token or a non-oauth token was used"""
+    @property
+    def shadow_banned(self) -> bool:
+        """True if the user making this request has been shadow-banned."""
+    @property
+    def device_id(self) -> str | None:
+        """The device_id which was set at authentication time, or
+        None for appservices, guests, and tokens generated by the admin API"""
+    @property
+    def app_service_id(self) -> str | None:
+        """The ID of the AS requesting on behalf of the user, or None."""
+    @property
+    def authenticated_entity(self) -> str:
+        """The entity that authenticated when making the request.
+
+        This is different to the user_id when an admin user or the server is
+        "puppeting" the user."""
+    def serialize(self) -> JsonDict:
+        """Converts self to a type that can be serialized as JSON, and then
+        deserialized by `deserialize`"""
+    @staticmethod
+    def deserialize(input: JsonDict) -> Requester:
+        """Converts a dict that was produced by `serialize` back into a
+        Requester."""
diff --git a/synapse/types/__init__.py b/synapse/types/__init__.py
index 8537a63bde..f2b1c3e42b 100644
--- a/synapse/types/__init__.py
+++ b/synapse/types/__init__.py
@@ -61,6 +61,7 @@ from twisted.internet.interfaces import (
 )
 
 from synapse.api.errors import Codes, SynapseError
+from synapse.synapse_rust.types import Requester
 from synapse.util.cancellation import cancellable
 from synapse.util.stringutils import parse_and_validate_server_name
 
@@ -70,7 +71,6 @@ if TYPE_CHECKING:
     from synapse.appservice.api import ApplicationService
     from synapse.events import EventBase
     from synapse.storage.databases.main import DataStore, PurgeEventsStore
-    from synapse.storage.databases.main.appservice import ApplicationServiceWorkerStore
     from synapse.storage.util.id_generators import MultiWriterIdGenerator
 
 
@@ -138,82 +138,6 @@ class ISynapseReactor(
     """The interfaces necessary for Synapse to function."""
 
 
-@attr.s(frozen=True, slots=True, auto_attribs=True)
-class Requester:
-    """
-    Represents the user making a request
-
-    Attributes:
-        user:  id of the user making the request
-        access_token_id:  *ID* of the access token used for this request, or
-            None for appservices, guests, and tokens generated by the admin API
-        is_guest:  True if the user making this request is a guest user
-        shadow_banned:  True if the user making this request has been shadow-banned.
-        device_id:  device_id which was set at authentication time, or
-            None for appservices, guests, and tokens generated by the admin API
-        app_service:  the AS requesting on behalf of the user
-        authenticated_entity: The entity that authenticated when making the request.
-            This is different to the user_id when an admin user or the server is
-            "puppeting" the user.
-    """
-
-    user: "UserID"
-    access_token_id: int | None
-    is_guest: bool
-    scope: set[str]
-    shadow_banned: bool
-    device_id: str | None
-    app_service: Optional["ApplicationService"]
-    authenticated_entity: str
-
-    def serialize(self) -> dict[str, Any]:
-        """Converts self to a type that can be serialized as JSON, and then
-        deserialized by `deserialize`
-
-        Returns:
-            dict
-        """
-        return {
-            "user_id": self.user.to_string(),
-            "access_token_id": self.access_token_id,
-            "is_guest": self.is_guest,
-            "scope": list(self.scope),
-            "shadow_banned": self.shadow_banned,
-            "device_id": self.device_id,
-            "app_server_id": self.app_service.id if self.app_service else None,
-            "authenticated_entity": self.authenticated_entity,
-        }
-
-    @staticmethod
-    def deserialize(
-        store: "ApplicationServiceWorkerStore", input: dict[str, Any]
-    ) -> "Requester":
-        """Converts a dict that was produced by `serialize` back into a
-        Requester.
-
-        Args:
-            store: Used to convert AS ID to AS object
-            input: A dict produced by `serialize`
-
-        Returns:
-            Requester
-        """
-        appservice = None
-        if input["app_server_id"]:
-            appservice = store.get_app_service_by_id(input["app_server_id"])
-
-        return Requester(
-            user=UserID.from_string(input["user_id"]),
-            access_token_id=input["access_token_id"],
-            is_guest=input["is_guest"],
-            scope=set(input.get("scope", [])),
-            shadow_banned=input["shadow_banned"],
-            device_id=input["device_id"],
-            app_service=appservice,
-            authenticated_entity=input["authenticated_entity"],
-        )
-
-
 def create_requester(
     user_id: Union[str, "UserID"],
     access_token_id: int | None = None,
@@ -258,7 +182,7 @@ def create_requester(
         scope,
         shadow_banned,
         device_id,
-        app_service,
+        app_service.id if app_service else None,
         authenticated_entity,
     )
 
diff --git a/tests/api/test_auth.py b/tests/api/test_auth.py
index f7905ced7e..e7ee7eec91 100644
--- a/tests/api/test_auth.py
+++ b/tests/api/test_auth.py
@@ -108,6 +108,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
 
     def test_get_user_by_req_appservice_valid_token(self) -> None:
         app_service = Mock(
+            id="as_id",
             token="foobar",
             url="a_url",
             sender=self.test_user_id,
@@ -132,6 +133,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             sender=self.test_user_id.to_string(),
             ip_range_whitelist=IPSet(["192.168.0.0/16"]),
         )
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
         self.store.get_user_by_access_token = AsyncMock(return_value=None)
 
@@ -151,6 +153,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             sender=self.test_user_id,
             ip_range_whitelist=IPSet(["192.168.0.0/16"]),
         )
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
         self.store.get_user_by_access_token = AsyncMock(return_value=None)
 
@@ -179,6 +182,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
 
     def test_get_user_by_req_appservice_missing_token(self) -> None:
         app_service = Mock(token="foobar", url="a_url", sender=self.test_user_id)
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
         self.store.get_user_by_access_token = AsyncMock(return_value=None)
 
@@ -199,6 +203,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             ip_range_whitelist=None,
         )
         app_service.is_interested_in_user = Mock(return_value=True)
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
 
         class FakeUserInfo:
@@ -226,6 +231,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             ip_range_whitelist=None,
         )
         app_service.is_interested_in_user = Mock(return_value=False)
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
         self.store.get_user_by_access_token = AsyncMock(return_value=None)
 
@@ -251,6 +257,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             ip_range_whitelist=None,
         )
         app_service.is_interested_in_user = Mock(return_value=True)
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
         # This just needs to return a truth-y value.
         self.store.get_user_by_id = AsyncMock(return_value={"is_guest": False})
@@ -285,6 +292,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             ip_range_whitelist=None,
         )
         app_service.is_interested_in_user = Mock(return_value=True)
+        app_service.id = "as_id"
         self.store.get_app_service_by_token = Mock(return_value=app_service)
         # This just needs to return a truth-y value.
         self.store.get_user_by_id = AsyncMock(return_value={"is_guest": False})
@@ -457,7 +465,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             is_guest=False,
             scope=set(),
             shadow_banned=False,
-            app_service=appservice,
+            app_service_id=appservice.id,
             authenticated_entity="@appservice:server",
         )
         self.get_success(self.auth_blocking.check_auth_blocking(requester=requester))
@@ -488,7 +496,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
             is_guest=False,
             scope=set(),
             shadow_banned=False,
-            app_service=appservice,
+            app_service_id=appservice.id,
             authenticated_entity="@appservice:server",
         )
         self.get_failure(
diff --git a/tests/api/test_ratelimiting.py b/tests/api/test_ratelimiting.py
index 0ef537841d..3237c6d412 100644
--- a/tests/api/test_ratelimiting.py
+++ b/tests/api/test_ratelimiting.py
@@ -40,6 +40,9 @@ class TestRatelimiter(unittest.HomeserverTestCase):
             rate_limited=True,
             sender=UserID.from_string("@as:example.com"),
         )
+        # The ratelimiter now resolves the AS via get_app_service_by_id, so the
+        # appservice must be in the store's cache for the lookup to hit.
+        self.hs.get_datastores().main.services_cache.append(appservice)
         as_requester = create_requester("@user:example.com", app_service=appservice)
 
         limiter = Ratelimiter(
@@ -76,6 +79,9 @@ class TestRatelimiter(unittest.HomeserverTestCase):
             rate_limited=False,
             sender=UserID.from_string("@as:example.com"),
         )
+        # The ratelimiter now resolves the AS via get_app_service_by_id, so the
+        # appservice must be in the store's cache for the lookup to hit.
+        self.hs.get_datastores().main.services_cache.append(appservice)
         as_requester = create_requester("@user:example.com", app_service=appservice)
 
         limiter = Ratelimiter(
diff --git a/tests/rest/client/test_transactions.py b/tests/rest/client/test_transactions.py
index 31586a451f..d62f9dd4a5 100644
--- a/tests/rest/client/test_transactions.py
+++ b/tests/rest/client/test_transactions.py
@@ -52,7 +52,7 @@ class HttpTransactionCacheTestCase(unittest.TestCase):
         self.mock_request = Mock()
         self.mock_request.path = b"/foo/bar"
         self.mock_requester = Mock()
-        self.mock_requester.app_service = None
+        self.mock_requester.app_service_id = None
         self.mock_requester.is_guest = False
         self.mock_requester.access_token_id = 1234