diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py index 62b35b56fa2..0b952f95a87 100644 --- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py +++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py @@ -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, ) diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_user_fields.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_user_fields.py index 4f0a64fc640..05cb4c4b6dd 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_user_fields.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_user_fields.py @@ -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)