Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/pass_through_endpoints/llm_provider_handlers/cursor_passthrough_logging_handler.py: 16%

1""" 

2Cursor Cloud Agents API - Pass-through Logging Handler 

3 

4Transforms Cursor API responses into standardized logging payloads 

5so they appear cleanly in the LiteLLM Logs page. 

6""" 

7 

8from datetime import datetime 

9from typing import Final 

10 

11import httpx 

12 

13from litellm._logging import verbose_proxy_logger 

14from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj 

15from litellm.litellm_core_utils.litellm_logging import ( 

16 get_standard_logging_object_payload, 

17) 

18from litellm.proxy._types import PassThroughEndpointLoggingTypedDict 

19from litellm.types.utils import StandardPassThroughResponseObject 

20 

21CURSOR_AGENT_ENDPOINTS: Final[dict[str, str]] = { 

22 "POST /v0/agents": "cursor:agent:create", 

23 "GET /v0/agents": "cursor:agent:list", 

24 "POST /v0/agents/{id}/followup": "cursor:agent:followup", 

25 "POST /v0/agents/{id}/stop": "cursor:agent:stop", 

26 "DELETE /v0/agents/{id}": "cursor:agent:delete", 

27 "GET /v0/agents/{id}/conversation": "cursor:agent:conversation", 

28 "GET /v0/agents/{id}": "cursor:agent:status", 

29 "GET /v0/me": "cursor:account:info", 

30 "GET /v0/models": "cursor:models:list", 

31 "GET /v0/repositories": "cursor:repositories:list", 

32} 

33 

34 

35def _classify_cursor_request(method: str, path: str) -> str: 

36 """Classify a Cursor API request into a readable operation name.""" 

37 normalized: Final = path.rstrip("/") 

38 

39 for pattern, operation in CURSOR_AGENT_ENDPOINTS.items(): 

40 pat_method, pat_path = pattern.split(" ", 1) 

41 if method.upper() != pat_method: 

42 continue 

43 

44 pat_parts = pat_path.strip("/").split("/") 

45 req_parts = normalized.strip("/").split("/") 

46 

47 if len(pat_parts) != len(req_parts): 

48 continue 

49 

50 match = True 

51 for pp, rp in zip(pat_parts, req_parts): 

52 if pp.startswith("{") and pp.endswith("}"): 

53 continue 

54 if pp != rp: 

55 match = False 

56 break 

57 if match: 

58 return operation 

59 

60 return f"cursor:{method.lower()}:{normalized}" 

61 

62 

63class CursorPassthroughLoggingHandler: 

64 """Handles logging for Cursor Cloud Agents pass-through requests.""" 

65 

66 @staticmethod 

67 def cursor_passthrough_handler( 

68 httpx_response: httpx.Response, 

69 response_body: dict, 

70 logging_obj: LiteLLMLoggingObj, 

71 url_route: str, 

72 result: str, 

73 start_time: datetime, 

74 end_time: datetime, 

75 cache_hit: bool, 

76 request_body: dict, 

77 **kwargs, 

78 ) -> PassThroughEndpointLoggingTypedDict: 

79 """ 

80 Transform a Cursor API response into a standard logging payload. 

81 """ 

82 try: 

83 method: Final = httpx_response.request.method 

84 path: Final = httpx.URL(url_route).path 

85 operation: Final = _classify_cursor_request(method, path) 

86 

87 agent_id: Final = response_body.get("id", "") 

88 agent_name: Final = response_body.get("name", "") 

89 agent_status: Final = response_body.get("status", "") 

90 

91 model_name: Final = f"cursor/{operation}" 

92 

93 summary_parts: Final = [] 

94 if agent_id: 

95 summary_parts.append(f"id={agent_id}") 

96 if agent_name: 

97 summary_parts.append(f"name={agent_name}") 

98 if agent_status: 

99 summary_parts.append(f"status={agent_status}") 

100 

101 response_summary: Final = ", ".join(summary_parts) if summary_parts else result 

102 

103 kwargs["model"] = model_name 

104 kwargs["response_cost"] = 0.0 

105 logging_obj.model_call_details["model"] = model_name 

106 logging_obj.model_call_details["custom_llm_provider"] = "cursor" 

107 logging_obj.model_call_details["response_cost"] = 0.0 

108 

109 standard_logging_object: Final = get_standard_logging_object_payload( 

110 kwargs=kwargs, 

111 init_response_obj=StandardPassThroughResponseObject(response=response_summary), 

112 start_time=start_time, 

113 end_time=end_time, 

114 logging_obj=logging_obj, 

115 status="success", 

116 ) 

117 kwargs["standard_logging_object"] = standard_logging_object 

118 

119 verbose_proxy_logger.debug( 

120 "Cursor passthrough logging: operation=%s, agent_id=%s", 

121 operation, 

122 agent_id, 

123 ) 

124 

125 return { 

126 "result": StandardPassThroughResponseObject(response=response_summary), 

127 "kwargs": kwargs, 

128 } 

129 except Exception as e: 

130 verbose_proxy_logger.exception("Error in Cursor passthrough logging handler: %s", e) 

131 return { 

132 "result": StandardPassThroughResponseObject(response=result), 

133 "kwargs": kwargs, 

134 }