refactor(proxy): type fresh Moyai connect helpers and drop MCP rpm getattr (#45315)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -3267,7 +3267,7 @@ class MCPServerManager:
             or "rfc8693",
             timeout=getattr(mcp_server, "timeout", None),
             max_concurrent_requests=getattr(mcp_server, "max_concurrent_requests", None),
-            rpm=getattr(mcp_server, "rpm", None),
+            rpm=mcp_server.rpm,
         )
         _warn_legacy_delegate_auth_if_applicable(new_server, source="database")
         if register_oauth_discovery:

litellm/proxy/moyai_endpoints.py

@@ -14,6 +14,8 @@ import json
 import os
 import secrets
 import time
+from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Annotated, Final
 from urllib.parse import urlencode, urlparse
 
@@ -63,6 +65,14 @@ class MoyaiConnectExchangeResponse(BaseModel):
     api_base: str
 
 
+@dataclass(frozen=True, slots=True)
+class _MoyaiConnectCode:
+    moyai_origin: str
+    nonce: str
+    exp: int
+    user_id: str | None
+
+
 def _b64url(data: bytes) -> str:
     return base64.urlsafe_b64encode(data).decode().rstrip("=")
 
@@ -104,7 +114,7 @@ def _sign_connect_code(master_key: str, moyai_url: str, user_id: str | None) ->
     return f"{_b64url(payload)}.{_b64url(signature)}"
 
 
-def _decode_connect_code(master_key: str, code: str) -> dict:
+def _decode_connect_code(master_key: str, code: str) -> _MoyaiConnectCode:
     try:
         payload_b64, signature_b64 = code.split(".", 1)
         payload_raw: Final = _b64url_decode(payload_b64)
@@ -122,7 +132,13 @@ def _decode_connect_code(master_key: str, code: str) -> dict:
         raise HTTPException(status_code=400, detail="Invalid Moyai connect code")
     if not isinstance(payload.get("moyai_origin"), str) or not isinstance(payload.get("nonce"), str):
         raise HTTPException(status_code=400, detail="Invalid Moyai connect code")
-    return payload
+    user_id: Final = payload.get("user_id")
+    return _MoyaiConnectCode(
+        moyai_origin=payload["moyai_origin"],
+        nonce=payload["nonce"],
+        exp=payload["exp"],
+        user_id=user_id if isinstance(user_id, str) else None,
+    )
 
 
 @router.post(
@@ -178,7 +194,7 @@ async def _claim_connect_nonce(prisma_client: "PrismaClient", nonce: str, exp: i
         raise HTTPException(status_code=400, detail="Invalid Moyai connect code")
 
 
-async def _moyai_key_alias(prisma_client, moyai_url: str) -> str:
+async def _moyai_key_alias(prisma_client: "PrismaClient", moyai_url: str) -> str:
     from litellm.repositories.verification_token_repository import (
         VerificationTokenRepository,
     )
@@ -192,16 +208,14 @@ async def _moyai_key_alias(prisma_client, moyai_url: str) -> str:
 
 
 @with_service_target(CONFIG_PARAMS_TARGET)
-async def _persist_moyai_url(prisma_client, moyai_url: str) -> None:
+async def _persist_moyai_url(prisma_client: "PrismaClient", moyai_url: str) -> None:
     from litellm.proxy.proxy_server import user_api_key_cache
 
-    existing: dict = {}
     db_existing: Final = await _ui_settings_db(UISettingsRepository(prisma_client)).find_unique(
         where={"id": "ui_settings"}
     )
-    if db_existing and db_existing.ui_settings:
-        raw: Final = db_existing.ui_settings
-        existing = json.loads(raw) if isinstance(raw, str) else dict(raw)
+    raw: Final = db_existing.ui_settings if db_existing else None
+    existing: Final[Mapping[str, object]] = (json.loads(raw) if isinstance(raw, str) else dict(raw)) if raw else {}
 
     ui_settings: Final = {**existing, "moyai_url": moyai_url}
     await _ui_settings_db(UISettingsRepository(prisma_client)).upsert(
@@ -232,13 +246,13 @@ async def moyai_connect_exchange(request: Request, body: MoyaiConnectExchangeReq
         moyai_url: Final = normalize_moyai_url(body.moyai_url)
     except ValueError:
         raise HTTPException(status_code=400, detail="Invalid Moyai connect code")
-    if moyai_url is None or _origin(moyai_url) != payload["moyai_origin"]:
+    if moyai_url is None or _origin(moyai_url) != payload.moyai_origin:
         raise HTTPException(status_code=400, detail="Invalid Moyai connect code")
 
     if prisma_client is None:
         raise HTTPException(status_code=400, detail="Moyai quick connect needs a database connected to the proxy")
 
-    await _claim_connect_nonce(prisma_client, payload["nonce"], payload["exp"])
+    await _claim_connect_nonce(prisma_client, payload.nonce, payload.exp)
 
     alias: Final = await _moyai_key_alias(prisma_client, moyai_url)
     key_response: Final = await generate_key_helper_fn(
@@ -248,7 +262,7 @@ async def moyai_connect_exchange(request: Request, body: MoyaiConnectExchangeReq
         metadata={
             "created_via": "moyai_quick_connect",
             "moyai_url": moyai_url,
-            "connected_by": payload.get("user_id"),
+            "connected_by": payload.user_id,
         },
         table_name="key",
         llm_router=llm_router,