audio streaming app
plyr.fm
refactor: extract main.py into focused modules (#890)
main.py was a 372-line monolith mixing middleware, observability config,
user-agent parsing, and endpoint definitions with app wiring. Now it's
138 lines of pure orchestration — imports, lifespan, and wiring.
- utilities/middleware.py: SecurityHeadersMiddleware
- utilities/observability.py: logfire setup, UA parsing, span enrichment
- api/meta.py: health, config, OAuth metadata, JWKS, robots, sitemap
- __main__.py: `python -m backend` convenience runner
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
authored by
Claude Opus 4.6
and committed by
GitHub
bfbb4cef
14e72933
+5
backend/src/backend/__main__.py
+5
backend/src/backend/__main__.py
+2
backend/src/backend/api/__init__.py
+2
backend/src/backend/api/__init__.py
···
2
2
3
3
from backend.api.account import router as account_router
4
4
from backend.api.artists import router as artists_router
5
+
from backend.api.meta import router as meta_router
5
6
from backend.api.audio import router as audio_router
6
7
from backend.api.auth import router as auth_router
7
8
from backend.api.exports import router as exports_router
···
22
23
"audio_router",
23
24
"auth_router",
24
25
"exports_router",
26
+
"meta_router",
25
27
"moderation_router",
26
28
"now_playing_router",
27
29
"oembed_router",
+149
backend/src/backend/api/meta.py
+149
backend/src/backend/api/meta.py
···
1
+
"""meta endpoints — health, config, OAuth metadata, robots, sitemap."""
2
+
3
+
from typing import Annotated, Any
4
+
5
+
from fastapi import APIRouter, Depends, HTTPException
6
+
from fastapi.responses import PlainTextResponse
7
+
from sqlalchemy import select
8
+
from sqlalchemy.ext.asyncio import AsyncSession
9
+
10
+
from backend._internal.auth import get_public_jwks, is_confidential_client
11
+
from backend.config import settings
12
+
from backend.models import Album, Artist, Track, get_db
13
+
14
+
router = APIRouter(tags=["meta"])
15
+
16
+
17
+
@router.get("/health")
18
+
async def health() -> dict[str, str]:
19
+
"""health check endpoint."""
20
+
return {"status": "ok"}
21
+
22
+
23
+
@router.get("/config")
24
+
async def get_public_config() -> dict[str, int | str | list[str]]:
25
+
"""expose public configuration to frontend."""
26
+
from backend.utilities.tags import DEFAULT_HIDDEN_TAGS
27
+
28
+
return {
29
+
"max_upload_size_mb": settings.storage.max_upload_size_mb,
30
+
"max_image_size_mb": 20, # hardcoded limit for cover art
31
+
"default_hidden_tags": DEFAULT_HIDDEN_TAGS,
32
+
"bufo_exclude_patterns": list(settings.bufo.exclude_patterns),
33
+
"bufo_include_patterns": list(settings.bufo.include_patterns),
34
+
"contact_email": settings.legal.contact_email,
35
+
"privacy_email": settings.legal.resolved_privacy_email,
36
+
"dmca_email": settings.legal.resolved_dmca_email,
37
+
"dmca_registration_number": settings.legal.dmca_registration_number,
38
+
"terms_last_updated": settings.legal.terms_last_updated.isoformat(),
39
+
}
40
+
41
+
42
+
@router.get("/oauth-client-metadata.json")
43
+
async def client_metadata() -> dict[str, Any]:
44
+
"""serve OAuth client metadata.
45
+
46
+
returns metadata for public or confidential client depending on
47
+
whether OAUTH_JWK is configured.
48
+
"""
49
+
# extract base URL from client_id for client_uri
50
+
client_uri = settings.atproto.client_id.replace("/oauth-client-metadata.json", "")
51
+
52
+
metadata: dict[str, Any] = {
53
+
"client_id": settings.atproto.client_id,
54
+
"client_name": settings.app.name,
55
+
"client_uri": client_uri,
56
+
"redirect_uris": [settings.atproto.redirect_uri],
57
+
"scope": settings.atproto.resolved_scope_with_teal(
58
+
settings.teal.play_collection, settings.teal.status_collection
59
+
),
60
+
"grant_types": ["authorization_code", "refresh_token"],
61
+
"response_types": ["code"],
62
+
"application_type": "web",
63
+
"dpop_bound_access_tokens": True,
64
+
}
65
+
66
+
if is_confidential_client():
67
+
# confidential client: use private_key_jwt authentication
68
+
# this gives us 180-day refresh tokens instead of 2-week
69
+
metadata["token_endpoint_auth_method"] = "private_key_jwt"
70
+
metadata["token_endpoint_auth_signing_alg"] = "ES256"
71
+
metadata["jwks_uri"] = f"{client_uri}/.well-known/jwks.json"
72
+
else:
73
+
# public client: no authentication
74
+
metadata["token_endpoint_auth_method"] = "none"
75
+
76
+
return metadata
77
+
78
+
79
+
@router.get("/.well-known/jwks.json")
80
+
async def jwks_endpoint() -> dict[str, Any]:
81
+
"""serve public JWKS for confidential client authentication.
82
+
83
+
returns 404 if confidential client is not configured.
84
+
"""
85
+
jwks = get_public_jwks()
86
+
if jwks is None:
87
+
raise HTTPException(
88
+
status_code=404,
89
+
detail="JWKS not available - confidential client not configured",
90
+
)
91
+
92
+
return jwks
93
+
94
+
95
+
@router.get("/robots.txt", include_in_schema=False)
96
+
async def robots_txt():
97
+
"""serve robots.txt to tell crawlers this is an API, not a website."""
98
+
return PlainTextResponse(
99
+
"User-agent: *\nDisallow: /\n",
100
+
media_type="text/plain",
101
+
)
102
+
103
+
104
+
@router.get("/sitemap-data")
105
+
async def sitemap_data(
106
+
db: Annotated[AsyncSession, Depends(get_db)],
107
+
) -> dict[str, Any]:
108
+
"""return minimal data needed to generate sitemap.xml.
109
+
110
+
returns tracks, artists, and albums with just IDs/slugs and timestamps.
111
+
the frontend renders this into XML at /sitemap.xml.
112
+
"""
113
+
# fetch all tracks (id, created_at)
114
+
tracks_result = await db.execute(
115
+
select(Track.id, Track.created_at).order_by(Track.created_at.desc())
116
+
)
117
+
tracks = [
118
+
{"id": row.id, "updated": row.created_at.strftime("%Y-%m-%d")}
119
+
for row in tracks_result.all()
120
+
]
121
+
122
+
# fetch all artists with at least one track (handle, updated_at)
123
+
artists_result = await db.execute(
124
+
select(Artist.handle, Artist.updated_at)
125
+
.join(Track, Artist.did == Track.artist_did)
126
+
.distinct()
127
+
.order_by(Artist.updated_at.desc())
128
+
)
129
+
artists = [
130
+
{"handle": row.handle, "updated": row.updated_at.strftime("%Y-%m-%d")}
131
+
for row in artists_result.all()
132
+
]
133
+
134
+
# fetch all albums (artist handle, slug, updated_at)
135
+
albums_result = await db.execute(
136
+
select(Album.slug, Artist.handle, Album.updated_at)
137
+
.join(Artist, Album.artist_did == Artist.did)
138
+
.order_by(Album.updated_at.desc())
139
+
)
140
+
albums = [
141
+
{
142
+
"handle": row.handle,
143
+
"slug": row.slug,
144
+
"updated": row.updated_at.strftime("%Y-%m-%d"),
145
+
}
146
+
for row in albums_result.all()
147
+
]
148
+
149
+
return {"tracks": tracks, "artists": artists, "albums": albums}
+17
-251
backend/src/backend/main.py
+17
-251
backend/src/backend/main.py
···
1
-
"""relay fastapi application."""
1
+
"""plyr.fm backend application."""
2
2
3
3
import asyncio
4
4
import logging
5
-
import re
6
-
import warnings
7
5
from collections.abc import AsyncIterator
8
6
from contextlib import asynccontextmanager
9
-
from typing import Annotated, Any
10
7
11
-
from fastapi import Depends, FastAPI, HTTPException, Request, Response, WebSocket
8
+
from fastapi import FastAPI, Request, Response
12
9
from fastapi.middleware.cors import CORSMiddleware
13
-
from fastapi.responses import ORJSONResponse, PlainTextResponse
10
+
from fastapi.responses import ORJSONResponse
14
11
from slowapi import _rate_limit_exceeded_handler
15
12
from slowapi.errors import RateLimitExceeded
16
13
from slowapi.middleware import SlowAPIMiddleware
17
-
from sqlalchemy import select
18
-
from sqlalchemy.ext.asyncio import AsyncSession
19
-
from starlette.middleware.base import BaseHTTPMiddleware
20
-
21
-
# filter pydantic warning from atproto library
22
-
warnings.filterwarnings(
23
-
"ignore",
24
-
message="The 'default' attribute with value None was provided to the `Field\\(\\)` function",
25
-
category=UserWarning,
26
-
module="pydantic._internal._generate_schema",
27
-
)
28
14
29
15
from backend._internal import notification_service, queue_service
30
-
from backend._internal.auth import get_public_jwks, is_confidential_client
31
16
from backend._internal.background import background_worker_lifespan
32
17
from backend.api import (
33
18
account_router,
···
35
20
audio_router,
36
21
auth_router,
37
22
exports_router,
23
+
meta_router,
38
24
moderation_router,
39
25
now_playing_router,
40
26
oembed_router,
···
50
36
from backend.api.lists import router as lists_router
51
37
from backend.api.migration import router as migration_router
52
38
from backend.config import settings
53
-
from backend.models import Album, Artist, Track, get_db
39
+
from backend.utilities.middleware import SecurityHeadersMiddleware
40
+
from backend.utilities.observability import (
41
+
configure_observability,
42
+
request_attributes_mapper,
43
+
suppress_warnings,
44
+
)
54
45
from backend.utilities.rate_limit import limiter
55
46
56
-
# configure logfire if enabled
57
-
if settings.observability.enabled:
58
-
import logfire
47
+
# suppress pydantic warnings before atproto imports
48
+
suppress_warnings()
59
49
60
-
if not settings.observability.write_token:
61
-
raise ValueError("LOGFIRE_WRITE_TOKEN must be set when LOGFIRE_ENABLED is true")
62
-
63
-
logfire.configure(
64
-
token=settings.observability.write_token,
65
-
environment=settings.observability.environment,
66
-
)
67
-
68
-
# configure logging with logfire handler
69
-
logging.basicConfig(
70
-
level=logging.DEBUG if settings.app.debug else logging.INFO,
71
-
handlers=[logfire.LogfireLoggingHandler()],
72
-
)
73
-
else:
74
-
logfire = None
75
-
# fallback to basic logging when logfire is disabled
76
-
logging.basicConfig(
77
-
level=logging.DEBUG if settings.app.debug else logging.INFO,
78
-
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
79
-
)
80
-
81
-
# reduce noise from verbose loggers
82
-
for logger_name in settings.observability.suppressed_loggers:
83
-
logging.getLogger(logger_name).setLevel(logging.WARNING)
50
+
# configure logfire + logging
51
+
logfire = configure_observability(settings)
84
52
85
53
logger = logging.getLogger(__name__)
86
54
87
-
# pattern to match plyrfm SDK/MCP user-agent headers
88
-
# format: "plyrfm/{version}" or "plyrfm-mcp/{version}"
89
-
_PLYRFM_UA_PATTERN = re.compile(r"^plyrfm(-mcp)?/(\d+\.\d+\.\d+)")
90
-
91
-
92
-
def parse_plyrfm_user_agent(user_agent: str | None) -> dict[str, str]:
93
-
"""parse plyrfm SDK/MCP user-agent into span attributes.
94
-
95
-
returns dict with:
96
-
- client_type: "sdk", "mcp", or "browser"
97
-
- client_version: version string (only for sdk/mcp)
98
-
"""
99
-
if not user_agent:
100
-
return {"client_type": "browser"}
101
-
102
-
match = _PLYRFM_UA_PATTERN.match(user_agent)
103
-
if not match:
104
-
return {"client_type": "browser"}
105
-
106
-
is_mcp = match.group(1) is not None # "-mcp" suffix present
107
-
version = match.group(2)
108
-
109
-
return {
110
-
"client_type": "mcp" if is_mcp else "sdk",
111
-
"client_version": version,
112
-
}
113
-
114
-
115
-
def request_attributes_mapper(
116
-
request: Request | WebSocket, attributes: dict[str, Any], /
117
-
) -> dict[str, Any] | None:
118
-
"""extract client metadata from request headers for span enrichment."""
119
-
user_agent = request.headers.get("user-agent")
120
-
return parse_plyrfm_user_agent(user_agent)
121
-
122
-
123
-
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
124
-
"""middleware to add security headers to all responses."""
125
-
126
-
async def dispatch(self, request: Request, call_next):
127
-
"""dispatch the request."""
128
-
response = await call_next(request)
129
-
130
-
# prevent MIME sniffing
131
-
response.headers["X-Content-Type-Options"] = "nosniff"
132
-
133
-
# prevent clickjacking
134
-
response.headers["X-Frame-Options"] = "DENY"
135
-
136
-
# enable browser XSS protection
137
-
response.headers["X-XSS-Protection"] = "1; mode=block"
138
-
139
-
# control referrer information
140
-
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
141
-
142
-
# enforce HTTPS in production (HSTS)
143
-
# skip in debug mode (localhost usually doesn't have https)
144
-
if not settings.app.debug:
145
-
response.headers["Strict-Transport-Security"] = (
146
-
"max-age=31536000; includeSubDomains"
147
-
)
148
-
149
-
return response
150
-
151
55
152
56
@asynccontextmanager
153
57
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
···
200
104
if logfire:
201
105
logfire.instrument_fastapi(app, request_attributes_mapper=request_attributes_mapper)
202
106
203
-
# add security headers middleware
107
+
# middleware
204
108
app.add_middleware(SecurityHeadersMiddleware) # type: ignore[arg-type]
205
-
206
-
# configure CORS - allow localhost for dev and cloudflare pages for production
207
109
app.add_middleware(
208
110
CORSMiddleware, # type: ignore[arg-type]
209
111
allow_origin_regex=settings.frontend.resolved_cors_origin_regex,
···
211
113
allow_methods=["*"],
212
114
allow_headers=["*"],
213
115
)
214
-
215
-
# add rate limiting middleware
216
116
app.add_middleware(SlowAPIMiddleware) # type: ignore[arg-type]
217
117
218
-
# include routers
118
+
# routers
219
119
app.include_router(auth_router)
220
120
app.include_router(account_router)
221
121
app.include_router(artists_router)
···
234
134
app.include_router(oembed_router)
235
135
app.include_router(stats_router)
236
136
app.include_router(users_router)
237
-
238
-
239
-
@app.get("/health")
240
-
async def health() -> dict[str, str]:
241
-
"""health check endpoint."""
242
-
return {"status": "ok"}
243
-
244
-
245
-
@app.get("/config")
246
-
async def get_public_config() -> dict[str, int | str | list[str]]:
247
-
"""expose public configuration to frontend."""
248
-
from backend.utilities.tags import DEFAULT_HIDDEN_TAGS
249
-
250
-
return {
251
-
"max_upload_size_mb": settings.storage.max_upload_size_mb,
252
-
"max_image_size_mb": 20, # hardcoded limit for cover art
253
-
"default_hidden_tags": DEFAULT_HIDDEN_TAGS,
254
-
"bufo_exclude_patterns": list(settings.bufo.exclude_patterns),
255
-
"bufo_include_patterns": list(settings.bufo.include_patterns),
256
-
"contact_email": settings.legal.contact_email,
257
-
"privacy_email": settings.legal.resolved_privacy_email,
258
-
"dmca_email": settings.legal.resolved_dmca_email,
259
-
"dmca_registration_number": settings.legal.dmca_registration_number,
260
-
"terms_last_updated": settings.legal.terms_last_updated.isoformat(),
261
-
}
262
-
263
-
264
-
@app.get("/oauth-client-metadata.json")
265
-
async def client_metadata() -> dict[str, Any]:
266
-
"""serve OAuth client metadata.
267
-
268
-
returns metadata for public or confidential client depending on
269
-
whether OAUTH_JWK is configured.
270
-
"""
271
-
# extract base URL from client_id for client_uri
272
-
client_uri = settings.atproto.client_id.replace("/oauth-client-metadata.json", "")
273
-
274
-
metadata: dict[str, Any] = {
275
-
"client_id": settings.atproto.client_id,
276
-
"client_name": settings.app.name,
277
-
"client_uri": client_uri,
278
-
"redirect_uris": [settings.atproto.redirect_uri],
279
-
"scope": settings.atproto.resolved_scope_with_teal(
280
-
settings.teal.play_collection, settings.teal.status_collection
281
-
),
282
-
"grant_types": ["authorization_code", "refresh_token"],
283
-
"response_types": ["code"],
284
-
"application_type": "web",
285
-
"dpop_bound_access_tokens": True,
286
-
}
287
-
288
-
if is_confidential_client():
289
-
# confidential client: use private_key_jwt authentication
290
-
# this gives us 180-day refresh tokens instead of 2-week
291
-
metadata["token_endpoint_auth_method"] = "private_key_jwt"
292
-
metadata["token_endpoint_auth_signing_alg"] = "ES256"
293
-
metadata["jwks_uri"] = f"{client_uri}/.well-known/jwks.json"
294
-
else:
295
-
# public client: no authentication
296
-
metadata["token_endpoint_auth_method"] = "none"
297
-
298
-
return metadata
299
-
300
-
301
-
@app.get("/.well-known/jwks.json")
302
-
async def jwks_endpoint() -> dict[str, Any]:
303
-
"""serve public JWKS for confidential client authentication.
304
-
305
-
returns 404 if confidential client is not configured.
306
-
"""
307
-
jwks = get_public_jwks()
308
-
if jwks is None:
309
-
raise HTTPException(
310
-
status_code=404,
311
-
detail="JWKS not available - confidential client not configured",
312
-
)
313
-
314
-
return jwks
315
-
316
-
317
-
@app.get("/robots.txt", include_in_schema=False)
318
-
async def robots_txt():
319
-
"""serve robots.txt to tell crawlers this is an API, not a website."""
320
-
return PlainTextResponse(
321
-
"User-agent: *\nDisallow: /\n",
322
-
media_type="text/plain",
323
-
)
324
-
325
-
326
-
@app.get("/sitemap-data")
327
-
async def sitemap_data(
328
-
db: Annotated[AsyncSession, Depends(get_db)],
329
-
) -> dict[str, Any]:
330
-
"""return minimal data needed to generate sitemap.xml.
331
-
332
-
returns tracks, artists, and albums with just IDs/slugs and timestamps.
333
-
the frontend renders this into XML at /sitemap.xml.
334
-
"""
335
-
# fetch all tracks (id, created_at)
336
-
tracks_result = await db.execute(
337
-
select(Track.id, Track.created_at).order_by(Track.created_at.desc())
338
-
)
339
-
tracks = [
340
-
{"id": row.id, "updated": row.created_at.strftime("%Y-%m-%d")}
341
-
for row in tracks_result.all()
342
-
]
343
-
344
-
# fetch all artists with at least one track (handle, updated_at)
345
-
artists_result = await db.execute(
346
-
select(Artist.handle, Artist.updated_at)
347
-
.join(Track, Artist.did == Track.artist_did)
348
-
.distinct()
349
-
.order_by(Artist.updated_at.desc())
350
-
)
351
-
artists = [
352
-
{"handle": row.handle, "updated": row.updated_at.strftime("%Y-%m-%d")}
353
-
for row in artists_result.all()
354
-
]
355
-
356
-
# fetch all albums (artist handle, slug, updated_at)
357
-
albums_result = await db.execute(
358
-
select(Album.slug, Artist.handle, Album.updated_at)
359
-
.join(Artist, Album.artist_did == Artist.did)
360
-
.order_by(Album.updated_at.desc())
361
-
)
362
-
albums = [
363
-
{
364
-
"handle": row.handle,
365
-
"slug": row.slug,
366
-
"updated": row.updated_at.strftime("%Y-%m-%d"),
367
-
}
368
-
for row in albums_result.all()
369
-
]
370
-
371
-
return {"tracks": tracks, "artists": artists, "albums": albums}
137
+
app.include_router(meta_router)
+35
backend/src/backend/utilities/middleware.py
+35
backend/src/backend/utilities/middleware.py
···
1
+
"""HTTP middleware."""
2
+
3
+
from fastapi import Request
4
+
from starlette.middleware.base import BaseHTTPMiddleware
5
+
6
+
from backend.config import settings
7
+
8
+
9
+
class SecurityHeadersMiddleware(BaseHTTPMiddleware):
10
+
"""middleware to add security headers to all responses."""
11
+
12
+
async def dispatch(self, request: Request, call_next):
13
+
"""dispatch the request."""
14
+
response = await call_next(request)
15
+
16
+
# prevent MIME sniffing
17
+
response.headers["X-Content-Type-Options"] = "nosniff"
18
+
19
+
# prevent clickjacking
20
+
response.headers["X-Frame-Options"] = "DENY"
21
+
22
+
# enable browser XSS protection
23
+
response.headers["X-XSS-Protection"] = "1; mode=block"
24
+
25
+
# control referrer information
26
+
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
27
+
28
+
# enforce HTTPS in production (HSTS)
29
+
# skip in debug mode (localhost usually doesn't have https)
30
+
if not settings.app.debug:
31
+
response.headers["Strict-Transport-Security"] = (
32
+
"max-age=31536000; includeSubDomains"
33
+
)
34
+
35
+
return response
+91
backend/src/backend/utilities/observability.py
+91
backend/src/backend/utilities/observability.py
···
1
+
"""observability configuration and request instrumentation."""
2
+
3
+
import logging
4
+
import re
5
+
import warnings
6
+
from types import ModuleType
7
+
from typing import Any
8
+
9
+
from fastapi import Request, WebSocket
10
+
11
+
from backend.config import Settings
12
+
13
+
# pattern to match plyrfm SDK/MCP user-agent headers
14
+
# format: "plyrfm/{version}" or "plyrfm-mcp/{version}"
15
+
_PLYRFM_UA_PATTERN = re.compile(r"^plyrfm(-mcp)?/(\d+\.\d+\.\d+)")
16
+
17
+
18
+
def suppress_warnings() -> None:
19
+
"""filter pydantic warnings emitted by the atproto library."""
20
+
warnings.filterwarnings(
21
+
"ignore",
22
+
message="The 'default' attribute with value None was provided to the `Field\\(\\)` function",
23
+
category=UserWarning,
24
+
module="pydantic._internal._generate_schema",
25
+
)
26
+
27
+
28
+
def configure_observability(settings: Settings) -> ModuleType | None:
29
+
"""configure logfire and logging. returns the logfire module if enabled, else None."""
30
+
if settings.observability.enabled:
31
+
import logfire
32
+
33
+
if not settings.observability.write_token:
34
+
raise ValueError(
35
+
"LOGFIRE_WRITE_TOKEN must be set when LOGFIRE_ENABLED is true"
36
+
)
37
+
38
+
logfire.configure(
39
+
token=settings.observability.write_token,
40
+
environment=settings.observability.environment,
41
+
)
42
+
43
+
# configure logging with logfire handler
44
+
logging.basicConfig(
45
+
level=logging.DEBUG if settings.app.debug else logging.INFO,
46
+
handlers=[logfire.LogfireLoggingHandler()],
47
+
)
48
+
else:
49
+
logfire = None
50
+
# fallback to basic logging when logfire is disabled
51
+
logging.basicConfig(
52
+
level=logging.DEBUG if settings.app.debug else logging.INFO,
53
+
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
54
+
)
55
+
56
+
# reduce noise from verbose loggers
57
+
for logger_name in settings.observability.suppressed_loggers:
58
+
logging.getLogger(logger_name).setLevel(logging.WARNING)
59
+
60
+
return logfire
61
+
62
+
63
+
def parse_plyrfm_user_agent(user_agent: str | None) -> dict[str, str]:
64
+
"""parse plyrfm SDK/MCP user-agent into span attributes.
65
+
66
+
returns dict with:
67
+
- client_type: "sdk", "mcp", or "browser"
68
+
- client_version: version string (only for sdk/mcp)
69
+
"""
70
+
if not user_agent:
71
+
return {"client_type": "browser"}
72
+
73
+
match = _PLYRFM_UA_PATTERN.match(user_agent)
74
+
if not match:
75
+
return {"client_type": "browser"}
76
+
77
+
is_mcp = match.group(1) is not None # "-mcp" suffix present
78
+
version = match.group(2)
79
+
80
+
return {
81
+
"client_type": "mcp" if is_mcp else "sdk",
82
+
"client_version": version,
83
+
}
84
+
85
+
86
+
def request_attributes_mapper(
87
+
request: Request | WebSocket, attributes: dict[str, Any], /
88
+
) -> dict[str, Any] | None:
89
+
"""extract client metadata from request headers for span enrichment."""
90
+
user_agent = request.headers.get("user-agent")
91
+
return parse_plyrfm_user_agent(user_agent)