"""Send prompts to Cursor IDE via cursor-agent CLI.
Requires the UNSANDBOXED_EXEC privilege or admin status.
Supports targeting arbitrary repository directories.
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import subprocess
from typing import TYPE_CHECKING
from tools.alter_privileges import has_privilege, PRIVILEGES
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "send_cursor_prompt"
TOOL_DESCRIPTION = (
"Send a prompt to Cursor IDE to modify or query a codebase. "
"Uses AI-powered code editing. Requires UNSANDBOXED_EXEC privilege. "
"Prompts should be specific and self-contained as the Cursor agent is stateless. "
"Defaults to the bot's own codebase; specify a directory to target a different repository."
)
AVAILABLE_MODELS = {
"composer-2": "CHEAP, FAST, PREFERABLE IN MOST CASES",
"composer-2-fast": "CHEAP, FAST, PREFERABLE IN MOST CASES",
"composer-1.5": "Recon – broad exploration and planning",
"opus-4.6-thinking": "Complex tasks requiring deep reasoning (VERY EXPENSIVE!!)",
"gpt-5.3-codex-spark-preview": "Moderate-complexity tasks (FREE)",
"sonnet-4.6-thinking": "Moderate-complexity tasks",
"grok": "Simple / quick tasks (may as well be free)",
}
DEFAULT_MODEL = "composer-2"
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": (
"The prompt describing what changes to make. Be specific and detailed, "
"in third-person perspective as if you are a human developer."
),
},
"model": {
"type": "string",
"enum": list(AVAILABLE_MODELS.keys()),
"description": (
"Which subagent model to use. Options: "
+ ", ".join(f"{k} ({v})" for k, v in AVAILABLE_MODELS.items())
+ f". Defaults to {DEFAULT_MODEL}."
),
},
"directory": {
"type": "string",
"description": (
"Absolute path to the repository/directory to run Cursor in. "
"Defaults to the bot's current working directory if omitted."
),
},
"timeout": {
"type": "integer",
"description": "Max seconds to wait for Cursor to respond (default 3600, max 7200).",
},
},
"required": ["prompt"],
}
[docs]
async def run(
prompt: str,
model: str = "",
directory: str = "",
timeout: int = 3600,
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
prompt (str): The prompt value.
model (str): Model to use for the subagent.
directory (str): The directory value.
timeout (int): Maximum wait time in seconds.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
user_id = getattr(ctx, "user_id", "") or ""
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if not await has_privilege(redis, user_id, PRIVILEGES["UNSANDBOXED_EXEC"], config):
return json.dumps({"success": False, "error": "The user does not have the UNSANDBOXED_EXEC privilege. Ask an admin to grant it with the alter_privileges tool."})
if not prompt or not prompt.strip():
return json.dumps({"success": False, "error": "Empty prompt provided."})
if timeout > 7200:
timeout = 7200
logger.warning("Timeout limited to 7200 seconds for user %s", user_id)
cwd = directory.strip() if directory and directory.strip() else os.getcwd()
if not os.path.isdir(cwd):
return json.dumps({"success": False, "error": f"Directory does not exist: {cwd}"})
selected_model = model.strip() if model and model.strip() else DEFAULT_MODEL
if selected_model not in AVAILABLE_MODELS:
return json.dumps({
"success": False,
"error": f"Unknown model '{selected_model}'. Choose from: {', '.join(AVAILABLE_MODELS)}",
})
logger.info("Executing cursor agent (%s) for user %s in %s: %s...", selected_model, user_id, cwd, prompt[:100])
try:
cmd_args = [
"cursor-agent",
"--api-key", "crsr_cc13d4b85df021a45f0d147b45784bf9285317816b227760510d130ebd49ff8b",
"--model", selected_model,
"--output-format", "json",
"--yolo",
"-p", prompt,
]
process = await asyncio.create_subprocess_exec(
*cmd_args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=cwd,
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
stdout_text = stdout.decode(errors="replace") if stdout else ""
stderr_text = stderr.decode(errors="replace") if stderr else ""
response_data = None
if stdout_text.strip():
try:
response_data = json.loads(stdout_text)
except json.JSONDecodeError:
response_data = {"raw_output": stdout_text}
result = {
"success": process.returncode == 0,
"return_code": process.returncode,
"working_directory": cwd,
"response": response_data,
}
if stderr_text.strip():
result["stderr"] = stderr_text.strip()
logger.info("Cursor agent completed for user %s, return code: %s", user_id, process.returncode)
return json.dumps(result, indent=2, ensure_ascii=False)
except asyncio.TimeoutError:
logger.warning("Cursor agent timed out for user %s: %s...", user_id, prompt[:50])
return json.dumps({
"success": False,
"error": f"Cursor agent timed out after {timeout} seconds.",
})
except FileNotFoundError:
logger.error("cursor-agent command not found")
return json.dumps({
"success": False,
"error": "cursor-agent command not found. Is Cursor CLI installed and in PATH?",
})
except Exception as e:
logger.error("Cursor agent error for user %s: %s", user_id, e, exc_info=True)
return json.dumps({"success": False, "error": f"An error occurred: {e}"})