feat: add Cursor Cloud Agents as a native pass-through provider

- Add CURSOR to LlmProviders enum
- Add /cursor/{endpoint:path} pass-through route with Basic Auth
- Add /cursor to mapped_pass_through_routes for proper routing
- Create CursorPassthroughLoggingHandler for Logs page visibility
  - Classifies operations (agent:create, agent:list, models:list, etc.)
  - Logs model as cursor/cursor:<operation> for clean Logs display
  - Tracks cost as $0 (subscription-based, no per-request pricing)
- Add Cursor to UI: provider enum, logo, credential fields
- Add provider_create_fields.json entry for LLM Credentials UI
- Add 18 unit tests covering route, auth, logging, and classification

Co-authored-by: Ishaan Jaff <ishaan-jaff@users.noreply.github.com>
This commit is contained in:
Cursor Agent 2026-02-28 19:40:34 +00:00
parent 6a916e7f55
commit 2044d565a6
10 changed files with 546 additions and 0 deletions

View file

@ -383,6 +383,7 @@ class LiteLLMRoutes(enum.Enum):
"/vertex-ai",
"/vertex_ai",
"/cohere",
"/cursor",
"/gemini",
"/anthropic",
"/langfuse",

View file

@ -2086,6 +2086,75 @@ class BaseOpenAIPassThroughHandler:
return joined_path_str
@router.api_route(
"/cursor/{endpoint:path}",
methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
tags=["Cursor Pass-through", "pass-through"],
)
async def cursor_proxy_route(
endpoint: str,
request: Request,
fastapi_response: Response,
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Pass-through endpoint for the Cursor Cloud Agents API.
Supports all Cursor Cloud Agents endpoints:
- GET /v0/agents List agents
- POST /v0/agents Launch an agent
- GET /v0/agents/{id} Agent status
- GET /v0/agents/{id}/conversation Agent conversation
- POST /v0/agents/{id}/followup Add follow-up
- POST /v0/agents/{id}/stop Stop an agent
- DELETE /v0/agents/{id} Delete an agent
- GET /v0/me API key info
- GET /v0/models List models
- GET /v0/repositories List GitHub repositories
Uses Basic Authentication (base64-encoded `API_KEY:`).
"""
import base64
base_target_url = os.getenv("CURSOR_API_BASE") or "https://api.cursor.com"
encoded_endpoint = httpx.URL(endpoint).path
if not encoded_endpoint.startswith("/"):
encoded_endpoint = "/" + encoded_endpoint
base_url = httpx.URL(base_target_url)
updated_url = base_url.copy_with(path=encoded_endpoint)
cursor_api_key = passthrough_endpoint_router.get_credentials(
custom_llm_provider="cursor",
region_name=None,
)
if cursor_api_key is None:
raise HTTPException(
status_code=401,
detail="Required 'CURSOR_API_KEY' in environment or credentials to make pass-through calls to Cursor.",
)
auth_value = base64.b64encode(
f"{cursor_api_key}:".encode("utf-8")
).decode("ascii")
endpoint_func = create_pass_through_route(
endpoint=endpoint,
target=str(updated_url),
custom_headers={"Authorization": f"Basic {auth_value}"},
custom_llm_provider="cursor",
)
received_value = await endpoint_func(
request,
fastapi_response,
user_api_key_dict,
)
return received_value
async def vertex_ai_live_websocket_passthrough(
websocket: WebSocket,
model: Optional[str] = None,

View file

@ -0,0 +1,141 @@
"""
Cursor Cloud Agents API - Pass-through Logging Handler
Transforms Cursor API responses into standardized logging payloads
so they appear cleanly in the LiteLLM Logs page.
"""
from datetime import datetime
from typing import Dict
import httpx
from litellm._logging import verbose_proxy_logger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.litellm_logging import (
get_standard_logging_object_payload,
)
from litellm.proxy._types import PassThroughEndpointLoggingTypedDict
from litellm.types.utils import StandardPassThroughResponseObject
CURSOR_AGENT_ENDPOINTS: Dict[str, str] = {
"POST /v0/agents": "cursor:agent:create",
"GET /v0/agents": "cursor:agent:list",
"POST /v0/agents/{id}/followup": "cursor:agent:followup",
"POST /v0/agents/{id}/stop": "cursor:agent:stop",
"DELETE /v0/agents/{id}": "cursor:agent:delete",
"GET /v0/agents/{id}/conversation": "cursor:agent:conversation",
"GET /v0/agents/{id}": "cursor:agent:status",
"GET /v0/me": "cursor:account:info",
"GET /v0/models": "cursor:models:list",
"GET /v0/repositories": "cursor:repositories:list",
}
def _classify_cursor_request(method: str, path: str) -> str:
"""Classify a Cursor API request into a readable operation name."""
normalized = path.rstrip("/")
for pattern, operation in CURSOR_AGENT_ENDPOINTS.items():
pat_method, pat_path = pattern.split(" ", 1)
if method.upper() != pat_method:
continue
pat_parts = pat_path.strip("/").split("/")
req_parts = normalized.strip("/").split("/")
if len(pat_parts) != len(req_parts):
continue
match = True
for pp, rp in zip(pat_parts, req_parts):
if pp.startswith("{") and pp.endswith("}"):
continue
if pp != rp:
match = False
break
if match:
return operation
return f"cursor:{method.lower()}:{normalized}"
class CursorPassthroughLoggingHandler:
"""Handles logging for Cursor Cloud Agents pass-through requests."""
@staticmethod
def cursor_passthrough_handler(
httpx_response: httpx.Response,
response_body: dict,
logging_obj: LiteLLMLoggingObj,
url_route: str,
result: str,
start_time: datetime,
end_time: datetime,
cache_hit: bool,
request_body: dict,
**kwargs,
) -> PassThroughEndpointLoggingTypedDict:
"""
Transform a Cursor API response into a standard logging payload.
"""
try:
method = httpx_response.request.method
path = httpx.URL(url_route).path
operation = _classify_cursor_request(method, path)
agent_id = response_body.get("id", "")
agent_name = response_body.get("name", "")
agent_status = response_body.get("status", "")
model_name = f"cursor/{operation}"
summary_parts = []
if agent_id:
summary_parts.append(f"id={agent_id}")
if agent_name:
summary_parts.append(f"name={agent_name}")
if agent_status:
summary_parts.append(f"status={agent_status}")
response_summary = ", ".join(summary_parts) if summary_parts else result
kwargs["model"] = model_name
kwargs["response_cost"] = 0.0
logging_obj.model_call_details["model"] = model_name
logging_obj.model_call_details["custom_llm_provider"] = "cursor"
logging_obj.model_call_details["response_cost"] = 0.0
standard_logging_object = get_standard_logging_object_payload(
kwargs=kwargs,
init_response_obj=StandardPassThroughResponseObject(
response=response_summary
),
start_time=start_time,
end_time=end_time,
logging_obj=logging_obj,
status="success",
)
kwargs["standard_logging_object"] = standard_logging_object
verbose_proxy_logger.debug(
"Cursor passthrough logging: operation=%s, agent_id=%s",
operation,
agent_id,
)
return {
"result": StandardPassThroughResponseObject(
response=response_summary
),
"kwargs": kwargs,
}
except Exception as e:
verbose_proxy_logger.exception(
"Error in Cursor passthrough logging handler: %s", e
)
return {
"result": StandardPassThroughResponseObject(response=result),
"kwargs": kwargs,
}

View file

@ -22,6 +22,9 @@ from .llm_provider_handlers.assembly_passthrough_logging_handler import (
from .llm_provider_handlers.cohere_passthrough_logging_handler import (
CoherePassthroughLoggingHandler,
)
from .llm_provider_handlers.cursor_passthrough_logging_handler import (
CursorPassthroughLoggingHandler,
)
from .llm_provider_handlers.gemini_passthrough_logging_handler import (
GeminiPassthroughLoggingHandler,
)
@ -60,6 +63,9 @@ class PassThroughEndpointLogging:
# Gemini
self.TRACKED_GEMINI_ROUTES = ["generateContent", "streamGenerateContent", "predictLongRunning"]
# Cursor Cloud Agents
self.TRACKED_CURSOR_ROUTES = ["/v0/agents", "/v0/me", "/v0/models", "/v0/repositories"]
# Vertex AI Live API WebSocket
self.TRACKED_VERTEX_AI_LIVE_ROUTES = ["/vertex_ai/live"]
@ -223,6 +229,25 @@ class PassThroughEndpointLogging:
)
kwargs = openai_passthrough_logging_handler_result["kwargs"]
elif self.is_cursor_route(url_route, custom_llm_provider):
cursor_passthrough_logging_handler_result = (
CursorPassthroughLoggingHandler.cursor_passthrough_handler(
httpx_response=httpx_response,
response_body=response_body or {},
logging_obj=logging_obj,
url_route=url_route,
result=result,
start_time=start_time,
end_time=end_time,
cache_hit=cache_hit,
request_body=request_body,
**kwargs,
)
)
standard_logging_response_object = (
cursor_passthrough_logging_handler_result["result"]
)
kwargs = cursor_passthrough_logging_handler_result["kwargs"]
elif self.is_vertex_ai_live_route(url_route):
from .llm_provider_handlers.vertex_ai_live_passthrough_logging_handler import (
VertexAILivePassthroughLoggingHandler,
@ -385,6 +410,22 @@ class PassThroughEndpointLogging:
return True
return False
def is_cursor_route(
self, url_route: str, custom_llm_provider: Optional[str] = None
):
"""Check if the URL route is a Cursor Cloud Agents API route."""
if custom_llm_provider == "cursor":
return True
parsed_url = urlparse(url_route)
if parsed_url.hostname and "api.cursor.com" in parsed_url.hostname:
return True
for route in self.TRACKED_CURSOR_ROUTES:
if route in url_route:
path = parsed_url.path if parsed_url.scheme else url_route
if path.startswith("/v0/"):
return custom_llm_provider == "cursor"
return False
def is_openai_route(self, url_route: str):
"""Check if the URL route is an OpenAI API route."""
if not url_route:

View file

@ -3025,5 +3025,33 @@
}
],
"default_model_placeholder": "gpt-3.5-turbo"
},
{
"provider": "CURSOR",
"provider_display_name": "Cursor",
"litellm_provider": "cursor",
"credential_fields": [
{
"key": "api_key",
"label": "Cursor API Key",
"placeholder": "Your Cursor API key from cursor.com/settings",
"tooltip": "Get your API key from the Cursor Dashboard at cursor.com/settings",
"required": true,
"field_type": "password",
"options": null,
"default_value": null
},
{
"key": "api_base",
"label": "API Base",
"placeholder": "https://api.cursor.com",
"tooltip": "Cursor API base URL (defaults to https://api.cursor.com)",
"required": false,
"field_type": "text",
"options": null,
"default_value": "https://api.cursor.com"
}
],
"default_model_placeholder": "cursor/claude-4-sonnet"
}
]

View file

@ -3185,6 +3185,7 @@ class LlmProviders(str, Enum):
CHUTES = "chutes"
XIAOMI_MIMO = "xiaomi_mimo"
LITELLM_AGENT = "litellm_agent"
CURSOR = "cursor"
# Create a set of all provider values for quick lookup

View file

@ -0,0 +1,116 @@
import os
import sys
from datetime import datetime
from unittest.mock import MagicMock
import httpx
import pytest
sys.path.insert(0, os.path.abspath("../../../../.."))
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.cursor_passthrough_logging_handler import (
CursorPassthroughLoggingHandler,
_classify_cursor_request,
)
from litellm.proxy.pass_through_endpoints.success_handler import (
PassThroughEndpointLogging,
)
class TestClassifyCursorRequest:
def test_should_classify_create_agent(self):
assert _classify_cursor_request("POST", "/v0/agents") == "cursor:agent:create"
def test_should_classify_list_agents(self):
assert _classify_cursor_request("GET", "/v0/agents") == "cursor:agent:list"
def test_should_classify_agent_status(self):
assert (
_classify_cursor_request("GET", "/v0/agents/bc_abc123")
== "cursor:agent:status"
)
def test_should_classify_agent_conversation(self):
assert (
_classify_cursor_request("GET", "/v0/agents/bc_abc123/conversation")
== "cursor:agent:conversation"
)
def test_should_classify_agent_followup(self):
assert (
_classify_cursor_request("POST", "/v0/agents/bc_abc123/followup")
== "cursor:agent:followup"
)
def test_should_classify_agent_stop(self):
assert (
_classify_cursor_request("POST", "/v0/agents/bc_abc123/stop")
== "cursor:agent:stop"
)
def test_should_classify_agent_delete(self):
assert (
_classify_cursor_request("DELETE", "/v0/agents/bc_abc123")
== "cursor:agent:delete"
)
def test_should_classify_me_endpoint(self):
assert _classify_cursor_request("GET", "/v0/me") == "cursor:account:info"
def test_should_classify_models_endpoint(self):
assert _classify_cursor_request("GET", "/v0/models") == "cursor:models:list"
def test_should_classify_repositories_endpoint(self):
assert (
_classify_cursor_request("GET", "/v0/repositories")
== "cursor:repositories:list"
)
class TestCursorRouteDetection:
def test_should_detect_cursor_route_by_custom_llm_provider(self):
handler = PassThroughEndpointLogging()
assert handler.is_cursor_route("https://api.cursor.com/v0/agents", "cursor")
def test_should_detect_cursor_route_by_hostname(self):
handler = PassThroughEndpointLogging()
assert handler.is_cursor_route("https://api.cursor.com/v0/agents")
def test_should_not_detect_non_cursor_route(self):
handler = PassThroughEndpointLogging()
assert not handler.is_cursor_route("https://api.openai.com/v1/completions")
class TestCursorPassthroughHandler:
def test_should_log_agent_creation_response(self):
mock_response = MagicMock(spec=httpx.Response)
mock_response.request = MagicMock()
mock_response.request.method = "POST"
mock_response.text = '{"id": "bc_abc123"}'
mock_logging_obj = MagicMock()
mock_logging_obj.litellm_call_id = "test-call-id"
mock_logging_obj.model_call_details = {}
response_body = {
"id": "bc_abc123",
"name": "Test Agent",
"status": "CREATING",
}
result = CursorPassthroughLoggingHandler.cursor_passthrough_handler(
httpx_response=mock_response,
response_body=response_body,
logging_obj=mock_logging_obj,
url_route="https://api.cursor.com/v0/agents",
result='{"id": "bc_abc123"}',
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body={"prompt": {"text": "Add README"}},
)
assert result["result"] is not None
assert result["kwargs"]["model"] == "cursor/cursor:agent:create"
assert result["kwargs"]["response_cost"] == 0.0
assert mock_logging_obj.model_call_details["custom_llm_provider"] == "cursor"

View file

@ -20,6 +20,7 @@ from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
RouteChecks,
bedrock_llm_proxy_route,
create_pass_through_route,
cursor_proxy_route,
llm_passthrough_factory_proxy_route,
milvus_proxy_route,
openai_proxy_route,
@ -2377,3 +2378,141 @@ class TestOpenAIPassthroughRoute:
# Verify result
assert result == {"id": "asst_123", "object": "assistant"}
class TestCursorProxyRoute:
"""Tests for the Cursor Cloud Agents pass-through route."""
@pytest.mark.asyncio
async def test_cursor_proxy_route_creates_pass_through_with_basic_auth(self):
"""should create a pass-through route with Basic Auth header for Cursor API"""
import base64
mock_request = MagicMock(spec=Request)
mock_request.method = "GET"
mock_request.query_params = {}
mock_request.headers = {}
mock_response = MagicMock(spec=Response)
mock_user_api_key_dict = MagicMock()
test_api_key = "test-cursor-api-key-123"
with patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
return_value=test_api_key,
), patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route"
) as mock_create_route:
mock_endpoint_func = AsyncMock(
return_value={"agents": [], "nextCursor": None}
)
mock_create_route.return_value = mock_endpoint_func
result = await cursor_proxy_route(
endpoint="v0/agents",
request=mock_request,
fastapi_response=mock_response,
user_api_key_dict=mock_user_api_key_dict,
)
mock_create_route.assert_called_once()
call_args = mock_create_route.call_args[1]
assert call_args["target"] == "https://api.cursor.com/v0/agents"
expected_auth = base64.b64encode(
f"{test_api_key}:".encode("utf-8")
).decode("ascii")
assert call_args["custom_headers"]["Authorization"] == f"Basic {expected_auth}"
assert result == {"agents": [], "nextCursor": None}
@pytest.mark.asyncio
async def test_cursor_proxy_route_raises_on_missing_api_key(self):
"""should raise 401 when no Cursor API key is available"""
mock_request = MagicMock(spec=Request)
mock_request.method = "POST"
mock_request.query_params = {}
mock_request.headers = {}
mock_response = MagicMock(spec=Response)
mock_user_api_key_dict = MagicMock()
with patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
return_value=None,
):
with pytest.raises(Exception) as exc_info:
await cursor_proxy_route(
endpoint="v0/agents",
request=mock_request,
fastapi_response=mock_response,
user_api_key_dict=mock_user_api_key_dict,
)
assert exc_info.value.status_code == 401
@pytest.mark.asyncio
async def test_cursor_proxy_route_custom_api_base(self):
"""should use CURSOR_API_BASE env var when set"""
mock_request = MagicMock(spec=Request)
mock_request.method = "GET"
mock_request.query_params = {}
mock_request.headers = {}
mock_response = MagicMock(spec=Response)
mock_user_api_key_dict = MagicMock()
with patch.dict(
os.environ, {"CURSOR_API_BASE": "https://custom-cursor.example.com"}
), patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
return_value="test-key",
), patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route"
) as mock_create_route:
mock_endpoint_func = AsyncMock(return_value={})
mock_create_route.return_value = mock_endpoint_func
await cursor_proxy_route(
endpoint="v0/me",
request=mock_request,
fastapi_response=mock_response,
user_api_key_dict=mock_user_api_key_dict,
)
call_args = mock_create_route.call_args[1]
assert call_args["target"] == "https://custom-cursor.example.com/v0/me"
@pytest.mark.asyncio
async def test_cursor_proxy_route_launch_agent(self):
"""should handle POST to launch an agent through the pass-through"""
mock_request = MagicMock(spec=Request)
mock_request.method = "POST"
mock_request.query_params = {}
mock_request.headers = {"content-type": "application/json"}
mock_response = MagicMock(spec=Response)
mock_user_api_key_dict = MagicMock()
with patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
return_value="test-key",
), patch(
"litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.create_pass_through_route"
) as mock_create_route:
mock_endpoint_func = AsyncMock(
return_value={
"id": "bc_abc123",
"name": "Test Agent",
"status": "CREATING",
}
)
mock_create_route.return_value = mock_endpoint_func
result = await cursor_proxy_route(
endpoint="v0/agents",
request=mock_request,
fastapi_response=mock_response,
user_api_key_dict=mock_user_api_key_dict,
)
call_args = mock_create_route.call_args[1]
assert call_args["target"] == "https://api.cursor.com/v0/agents"
assert result["id"] == "bc_abc123"
assert result["status"] == "CREATING"

View file

@ -0,0 +1,5 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 100 100" fill="none">
<rect width="100" height="100" rx="20" fill="#000000"/>
<path d="M25 75V25L70 50L25 75Z" fill="white"/>
<path d="M70 50L50 75H25L70 50Z" fill="#AAAAAA"/>
</svg>

After

Width:  |  Height:  |  Size: 242 B

View file

@ -9,6 +9,7 @@ export enum Providers {
Azure_AI_Studio = "Azure AI Foundry (Studio)",
Cerebras = "Cerebras",
Cohere = "Cohere",
Cursor = "Cursor",
Dashscope = "Dashscope",
Databricks = "Databricks (Qwen API)",
DeepInfra = "DeepInfra",
@ -60,6 +61,7 @@ export const provider_map: Record<string, string> = {
MiniMax: "minimax",
MistralAI: "mistral",
Cohere: "cohere",
Cursor: "cursor",
OpenAI_Compatible: "openai",
OpenAI_Text_Compatible: "text-completion-openai",
Vertex_AI: "vertex_ai",
@ -107,6 +109,7 @@ export const providerLogoMap: Record<string, string> = {
[Providers.SageMaker]: `${asset_logos_folder}bedrock.svg`,
[Providers.Cerebras]: `${asset_logos_folder}cerebras.svg`,
[Providers.Cohere]: `${asset_logos_folder}cohere.svg`,
[Providers.Cursor]: `${asset_logos_folder}cursor.svg`,
[Providers.Databricks]: `${asset_logos_folder}databricks.svg`,
[Providers.Dashscope]: `${asset_logos_folder}dashscope.svg`,
[Providers.Deepseek]: `${asset_logos_folder}deepseek.svg`,
@ -206,6 +209,8 @@ export const getPlaceholder = (selectedProvider: string): string => {
return "runwayml/gen4_turbo";
} else if (selectedProvider === Providers.Watsonx) {
return "watsonx/ibm/granite-3-3-8b-instruct";
} else if (selectedProvider === Providers.Cursor) {
return "cursor/claude-4-sonnet";
} else {
return "gpt-3.5-turbo";
}