fix(mcp): return 409 on BYOK/user-fields conflict; coerce Pydantic user_fields

- store_mcp_user_credential now catches the ValueError raised by
  store_user_credential when the (user, server) row already holds a
  user-fields payload, returning HTTP 409 instead of leaking a 500.
- coerce_user_fields now also accepts MCPUserField Pydantic instances
  (the shape used by LiteLLM_MCPServerTable.user_fields), so the
  management-layer annotation and enforcement paths no longer return
  silently empty when callers pass a parsed server record.
This commit is contained in:
mateo-berri 2026-05-19 07:19:18 +00:00
parent 6b0af7f89a
commit 5215fbeeeb
No known key found for this signature in database
2 changed files with 35 additions and 3 deletions

View file

@ -1918,9 +1918,19 @@ if MCP_AVAILABLE:
detail={"error": "User ID not found in token"},
)
if payload.save:
await store_user_credential(
prisma_client, user_id, server_id, payload.credential
)
try:
await store_user_credential(
prisma_client, user_id, server_id, payload.credential
)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={
"error": "credential_conflict",
"message": str(e),
"server_id": server_id,
},
)
from litellm.proxy._experimental.mcp_server.server import (
_invalidate_byok_cred_cache,
)

View file

@ -179,6 +179,28 @@ def test_coerce_user_fields_empty_when_missing():
assert server_has_user_fields(srv) is False
def test_coerce_user_fields_accepts_litellm_mcp_server_table():
"""LiteLLM_MCPServerTable.user_fields is List[MCPUserField] (Pydantic
instances), not List[dict]. The helper must normalise both shapes so
the management-layer annotation / enforcement paths don't silently
return empty results.
"""
table = LiteLLM_MCPServerTable(
server_id="s3",
transport=MCPTransport.http,
user_fields=[
{"field_key": "TOKEN", "header_name": "Authorization", "required": True},
{"field_key": "WS", "header_name": "X-Workspace", "required": False},
],
)
assert isinstance(table.user_fields[0], MCPUserField)
coerced = coerce_user_fields(table)
assert [f["field_key"] for f in coerced] == ["TOKEN", "WS"]
assert server_has_user_fields(table) is True
missing = compute_missing_user_fields(table, None)
assert [f["field_key"] for f in missing] == ["TOKEN"]
def test_compute_missing_required_only():
srv = _gmail_server()
missing = compute_missing_user_fields(srv, None)