refactor(config): migrate to yaml and forward auth (#3)

main
Yixing Lao 2026-04-24 18:42:59 +08:00 committed by GitHub
parent 1717331057
commit c238a40045
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 495 additions and 370 deletions

View File

@ -1,22 +0,0 @@
# Copy this file to ~/.deepseek-cursor-proxy/.env.
# The proxy loads that file automatically and keeps secrets out of the repo.
DEEPSEEK_API_KEY=sk-your-deepseek-key
# Use this as the OpenAI API key in Cursor.
PROXY_API_KEY=cursor-local-token
DEEPSEEK_MODEL=deepseek-v4-pro
DEEPSEEK_BASE_URL=https://api.deepseek.com
DEEPSEEK_THINKING=enabled
DEEPSEEK_REASONING_EFFORT=high
CURSOR_DISPLAY_REASONING=true
PROXY_HOST=127.0.0.1
PROXY_PORT=9000
PROXY_NGROK=true
PROXY_VERBOSE=false
PROXY_LOG_BODIES=false
# Optional. Default: ~/.deepseek-cursor-proxy/reasoning_content.sqlite3
REASONING_CONTENT_PATH=~/.deepseek-cursor-proxy/reasoning_content.sqlite3

152
README.md
View File

@ -1,98 +1,91 @@
# deepseek-cursor-proxy
A simple proxy that caches and restores DeepSeek `reasoning_content` across tool-call turns in Cursor, making thinking models like `deepseek-v4-pro` and `deepseek-v4-flash` work correctly.
Compatibility proxy connecting Cursor to DeepSeek thinking models (`deepseek-v4-pro` and `deepseek-v4-flash`).
## What It Does
- Caches DeepSeek `reasoning_content` from regular and streamed responses, then restores it on later tool-call turns when Cursor omits it.
- Mirrors streamed `reasoning_content` into Cursor-visible `<think>...</think>` text so thinking tokens are shown in Cursor BYOK/proxy chats. Cursor currently renders this as normal chat text, not as a native collapsible Thinking block.
- Provides other compatibility fixes for running Cursor with the DeepSeek official API.
- ✅ Caches DeepSeek `reasoning_content` from regular and streamed responses, then restores it on later tool-call turns when Cursor omits it. See [DeepSeek docs](https://api-docs.deepseek.com/guides/thinking_mode#tool-calls) for more details.
- ✅ Mirrors streamed `reasoning_content` into Cursor-visible `<think>...</think>` text so that thinking tokens are shown in Cursor's UI. For BYOK/proxy mode, Cursor renders this as normal text, not as a native collapsible thinking block.
- ✅ Starts an ngrok tunnel so Cursor can reach the local proxy.
- ✅ Provides other compatibility fixes to make DeepSeek models run well in Cursor.
## Why This Exists
DeepSeek thinking mode returns `reasoning_content` separately from final `content`. After an assistant turn with tool calls, DeepSeek requires that same `reasoning_content` to be sent back in later requests. Cursor can omit it in custom OpenAI-compatible flows, causing `The reasoning_content in the thinking mode must be passed back to the API.` This proxy caches reasoning by conversation prefix, message signature, and tool-call IDs, then restores it before forwarding to DeepSeek.
For streamed responses, the proxy also mirrors DeepSeek `reasoning_content` into Cursor-visible `<think>...</think>` content while leaving the original `reasoning_content` field intact. This lets Cursor display the thinking text in OpenAI-compatible BYOK/proxy flows, and the proxy strips those display-only tags from later assistant history before replaying it to DeepSeek.
This repo fixes the following error:
This repository fixes the following Cursor + DeepSeek tool-call error with thinking mode enabled:
![Error 400 - reasoning_content must be passed back](assets/error_400.png)
```txt
⚠️ Connection Error
Provider returned error: {"error":{"message":"The reasoning_content in the thinking mode must be passed back to the
API.","type":"invalid_request_error","param":null,"code":"invalid_request_error"}}
Provider returned error:
{
"error": {
"message": "The reasoning_content in the thinking mode must be passed back to the API.",
"type": "invalid_request_error",
"param": null,
"code": "invalid_request_error"
}
}
```
## 1. Install
## Usage
```bash
source ~/miniconda3/etc/profile.d/conda.sh
conda activate pytools
PIP_REQUIRE_VIRTUALENV=false python -m pip install -e .
```
### Step 1: Set Up ngrok
## 2. Configure
Create an ngrok account, visit ngrok's Dashboard: https://dashboard.ngrok.com
```bash
mkdir -p ~/.deepseek-cursor-proxy
chmod 700 ~/.deepseek-cursor-proxy
cp .env.example ~/.deepseek-cursor-proxy/.env
chmod 600 ~/.deepseek-cursor-proxy/.env
```
![ngrok dashboard showing the public URL](assets/ngrok_dashboard.png)
`.env.example` is only a safe template. The proxy loads `~/.deepseek-cursor-proxy/.env` automatically, and that file should stay outside this repository because it contains your keys.
Edit `~/.deepseek-cursor-proxy/.env`:
```bash
DEEPSEEK_API_KEY=sk-your-deepseek-key
PROXY_API_KEY=cursor-local-token
CURSOR_DISPLAY_REASONING=true
```
Keep `PROXY_API_KEY` set when using ngrok because the proxy will be reachable from the public internet.
By default, reasoning cache data is stored at:
```text
~/.deepseek-cursor-proxy/reasoning_content.sqlite3
```
Override it with `REASONING_CONTENT_PATH` or `deepseek-cursor-proxy --reasoning-content-path <path>` only when you need a custom location.
## 3. Set Up Ngrok Once
- Create/login to an ngrok account: https://dashboard.ngrok.com/signup
- Copy your authtoken from the dashboard: https://dashboard.ngrok.com/get-started/your-authtoken
Then, install and authenticate ngrok once:
```bash
brew install ngrok
ngrok config add-authtoken <your-ngrok-token>
```
## 4. Run
### Step 2: Add Cursor Custom Model
In Cursor, add the DeepSeek custom model and point it at this proxy:
- Model: `deepseek-v4-pro`
- API Key: your DeepSeek API key
- Base URL: your ngrok HTTPS URL with the `/v1` API version path
For example, if ngrok dashboard shows `https://example.ngrok-free.app`, use:
```text
https://example.ngrok-free.app/v1
```
![Cursor settings for DeepSeek through the proxy](assets/cursor_config.png)
Note: you can toggle the custom API on and off with:
- macOS: `Cmd+Shift+0`
- Windows/Linux: `Ctrl+Shift+0`
### Step 3: Start the Proxy Server
Install and run the proxy:
```bash
conda create -n dcp python=3.10 -y
conda activate dcp
pip install -e .
deepseek-cursor-proxy --verbose
```
The proxy prints a line like:
The proxy creates `~/.deepseek-cursor-proxy/config.yaml` on first run.
```text
Cursor Base URL: https://example.ngrok-free.app/v1
```
This will also print the ngrok public URL. If it differs from the one in Cursor, update it in Cursor's Base URL field.
Use that URL in Cursor. If you do not use ngrok and point Cursor at `localhost` or `127.0.0.1`, Cursor may fail with `ssrf_blocked: connection to private IP is blocked`.
### Step 4: Chat with DeepSeek in Cursor
## 5. Cursor Settings
Select `deepseek-v4-pro` in Cursor and use chat or agent mode as usual.
- OpenAI Base URL: the printed ngrok URL ending in `/v1`
- OpenAI API Key: the value of `PROXY_API_KEY`
- Model: `deepseek-v4-pro`
![Chatting with DeepSeek in Cursor](assets/cursor_chat.png)
## Useful Commands
## Debugging and Development
Run without ngrok for local curl testing:
@ -100,24 +93,10 @@ Run without ngrok for local curl testing:
PROXY_NGROK=false deepseek-cursor-proxy --port 9000 --verbose
```
Disable the Cursor display mirror if you only want raw OpenAI-compatible response fields:
Use another config file:
```bash
CURSOR_DISPLAY_REASONING=false deepseek-cursor-proxy --verbose
```
Log full request bodies only when needed:
```bash
deepseek-cursor-proxy --ngrok --verbose --log-bodies
```
This prints the Cursor request body, the normalized DeepSeek request body, DeepSeek error bodies, and the final streamed assistant message.
Use a different env file for development:
```bash
deepseek-cursor-proxy --config ./dev.env
deepseek-cursor-proxy --config ./dev.config.yaml
```
Run tests:
@ -125,24 +104,3 @@ Run tests:
```bash
PYTHONPATH=src python -m unittest discover -s tests
```
## Development
Pre-commit runs whitespace checks, Black, and Ruff:
```bash
PIP_REQUIRE_VIRTUALENV=false python -m pip install -e ".[dev]"
pre-commit install
pre-commit run --all-files
```
## Notes
- Distribution name: `deepseek-cursor-proxy`
- Import package: `deepseek_cursor_proxy`
- User config file: `~/.deepseek-cursor-proxy/.env`
- Cache file: `~/.deepseek-cursor-proxy/reasoning_content.sqlite3`
- DeepSeek thinking docs: https://api-docs.deepseek.com/guides/thinking_mode
- DeepSeek chat completion docs: https://api-docs.deepseek.com/api/create-chat-completion
- Cursor forum report: https://forum.cursor.com/t/compatibility-with-deepseek-models-design-to-return-reasoning-content-after-tool-calls/158905
- ngrok setup docs: https://ngrok.com/downloads

BIN
assets/cursor_chat.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 72 KiB

BIN
assets/cursor_config.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 74 KiB

BIN
assets/ngrok_dashboard.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 45 KiB

17
config.example.yaml Normal file
View File

@ -0,0 +1,17 @@
# This file was created automatically at ~/.deepseek-cursor-proxy/config.yaml.
# API keys are read from Cursor's Authorization header and forwarded upstream.
base_url: https://api.deepseek.com
model: deepseek-v4-pro
thinking: enabled
reasoning_effort: high
display_reasoning: true
host: 127.0.0.1
port: 9000
ngrok: true
verbose: false
log_bodies: false
request_timeout: 300
reasoning_content_path: reasoning_content.sqlite3

View File

@ -25,7 +25,9 @@ classifiers = [
"Topic :: Internet :: Proxy Servers",
"Topic :: Software Development",
]
dependencies = []
dependencies = [
"PyYAML>=6.0",
]
[project.optional-dependencies]
dev = [

View File

@ -3,15 +3,36 @@ from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
import os
import yaml
APP_DIR_NAME = ".deepseek-cursor-proxy"
CONFIG_FILE_NAME = ".env"
CONFIG_FILE_NAME = "config.yaml"
REASONING_CONTENT_FILE_NAME = "reasoning_content.sqlite3"
TRUE_VALUES = {"1", "true", "yes", "on"}
FALSE_VALUES = {"0", "false", "no", "off"}
MISSING = object()
DEFAULT_CONFIG_TEXT = """# This file was created automatically at ~/.deepseek-cursor-proxy/config.yaml.
# API keys are read from Cursor's Authorization header and forwarded upstream.
base_url: https://api.deepseek.com
model: deepseek-v4-pro
thinking: enabled
reasoning_effort: high
display_reasoning: true
host: 127.0.0.1
port: 9000
ngrok: true
verbose: false
log_bodies: false
request_timeout: 300
reasoning_content_path: reasoning_content.sqlite3
"""
def default_app_dir() -> Path:
@ -26,45 +47,65 @@ def default_reasoning_content_path() -> Path:
return default_app_dir() / REASONING_CONTENT_FILE_NAME
def load_env_file(env_file_path: str | Path) -> dict[str, str]:
env_file_path = Path(env_file_path)
if not env_file_path.exists():
def populate_default_config_file(config_path: Path) -> None:
config_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
config_path.parent.chmod(0o700)
config_path.write_text(DEFAULT_CONFIG_TEXT, encoding="utf-8")
config_path.chmod(0o600)
def load_config_file(config_path: str | Path) -> dict[str, Any]:
config_path = Path(config_path).expanduser()
if not config_path.exists():
return {}
values: dict[str, str] = {}
for raw_line in env_file_path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
if line.startswith("export "):
line = line.removeprefix("export ").strip()
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
if key:
values[key] = value
return values
try:
loaded = yaml.safe_load(config_path.read_text(encoding="utf-8"))
except yaml.YAMLError as exc:
raise ValueError(f"Invalid YAML config at {config_path}: {exc}") from exc
if loaded is None:
return {}
if not isinstance(loaded, Mapping):
raise ValueError(f"Config file must contain a YAML mapping: {config_path}")
return dict(loaded)
def merged_env(
env: Mapping[str, str] | None, env_file_path: str | Path | None
) -> dict[str, str]:
live_env = dict(os.environ if env is None else env)
config_path = Path(
env_file_path
def resolve_config_path(
env: Mapping[str, str] | None, config_path: str | Path | None
) -> Path:
live_env = os.environ if env is None else env
return Path(
config_path
or live_env.get("DEEPSEEK_CURSOR_PROXY_CONFIG_PATH")
or default_config_path()
)
values = load_env_file(config_path)
values.update(live_env)
return values
).expanduser()
def env_bool(values: Mapping[str, str], name: str, default: bool) -> bool:
value = values.get(name)
if value is None:
def setting_value(
settings: Mapping[str, Any],
env: Mapping[str, str],
key: str,
env_name: str,
) -> Any:
if env_name in env:
return env[env_name]
return settings.get(key, MISSING)
def as_str(value: Any, default: str) -> str:
if value is MISSING or value is None:
return default
normalized = value.strip().lower()
return str(value)
def as_bool(value: Any, default: bool) -> bool:
if value is MISSING or value is None:
return default
if isinstance(value, bool):
return value
if isinstance(value, int):
return bool(value)
normalized = str(value).strip().lower()
if normalized in TRUE_VALUES:
return True
if normalized in FALSE_VALUES:
@ -72,48 +113,45 @@ def env_bool(values: Mapping[str, str], name: str, default: bool) -> bool:
return default
def env_int(values: Mapping[str, str], name: str, default: int) -> int:
value = values.get(name)
if value is None:
def as_int(value: Any, default: int) -> int:
if value is MISSING or value is None:
return default
try:
return int(value)
except ValueError:
except (TypeError, ValueError):
return default
def env_float(values: Mapping[str, str], name: str, default: float) -> float:
value = values.get(name)
if value is None:
def as_float(value: Any, default: float) -> float:
if value is MISSING or value is None:
return default
try:
return float(value)
except ValueError:
except (TypeError, ValueError):
return default
def env_tuple(
values: Mapping[str, str], name: str, default: tuple[str, ...]
) -> tuple[str, ...]:
value = values.get(name)
if not value:
return default
return tuple(item.strip() for item in value.split(",") if item.strip())
def as_path(value: Any, default_path: Path, relative_base: Path) -> Path:
if value is MISSING or value is None or value == "":
return default_path
candidate_path = Path(str(value)).expanduser()
if candidate_path.is_absolute():
return candidate_path
return relative_base / candidate_path
def env_path(
values: Mapping[str, str],
names: tuple[str, ...],
default_path: Path,
) -> Path:
for env_name in names:
value = values.get(env_name)
if value:
candidate_path = Path(value).expanduser()
if candidate_path.is_absolute():
return candidate_path
return default_path.parent / candidate_path
return default_path
def settings_and_env(
env: Mapping[str, str] | None, config_path: str | Path | None
) -> tuple[dict[str, Any], dict[str, str], Path]:
live_env = dict(os.environ if env is None else env)
config_path = resolve_config_path(live_env, config_path)
if (
config_path == default_config_path()
and "DEEPSEEK_CURSOR_PROXY_CONFIG_PATH" not in live_env
and not config_path.exists()
):
populate_default_config_file(config_path)
return load_config_file(config_path), live_env, config_path
@dataclass(frozen=True)
@ -121,8 +159,6 @@ class ProxyConfig:
host: str = "127.0.0.1"
port: int = 9000
upstream_base_url: str = "https://api.deepseek.com"
upstream_api_key: str = ""
proxy_api_key: str | None = None
upstream_model: str = "deepseek-v4-pro"
allow_model_passthrough: bool = False
thinking: str = "enabled"
@ -133,52 +169,143 @@ class ProxyConfig:
verbose: bool = False
log_bodies: bool = False
ngrok: bool = False
model_list: tuple[str, ...] = ("deepseek-v4-pro", "deepseek-v4-flash")
@classmethod
def from_env(
def from_file(
cls: type[ProxyConfig],
env: Mapping[str, str] | None = None,
env_file_path: str | Path | None = None,
config_path: str | Path | None = None,
) -> "ProxyConfig":
values = merged_env(env, env_file_path)
thinking = values.get("DEEPSEEK_THINKING", "enabled").strip().lower()
settings, live_env, resolved_config_path = settings_and_env(env, config_path)
config_dir = resolved_config_path.parent
thinking = (
as_str(
setting_value(
settings,
live_env,
"thinking",
"DEEPSEEK_THINKING",
),
"enabled",
)
.strip()
.lower()
)
if thinking in {"passthrough", "pass-through", "pass_through"}:
thinking = "pass-through"
if thinking not in {"enabled", "disabled", "pass-through"}:
thinking = "enabled"
return cls(
host=values.get("PROXY_HOST", "127.0.0.1"),
port=env_int(values, "PROXY_PORT", 9000),
upstream_base_url=values.get(
"DEEPSEEK_BASE_URL", "https://api.deepseek.com"
host=as_str(
setting_value(
settings,
live_env,
"host",
"PROXY_HOST",
),
"127.0.0.1",
),
port=as_int(
setting_value(
settings,
live_env,
"port",
"PROXY_PORT",
),
9000,
),
upstream_base_url=as_str(
setting_value(
settings,
live_env,
"base_url",
"DEEPSEEK_BASE_URL",
),
"https://api.deepseek.com",
).rstrip("/"),
upstream_api_key=values.get("DEEPSEEK_API_KEY", ""),
proxy_api_key=values.get("PROXY_API_KEY") or None,
upstream_model=values.get("DEEPSEEK_MODEL", "deepseek-v4-pro"),
allow_model_passthrough=env_bool(
values, "DEEPSEEK_ALLOW_MODEL_PASSTHROUGH", False
upstream_model=as_str(
setting_value(
settings,
live_env,
"model",
"DEEPSEEK_MODEL",
),
"deepseek-v4-pro",
),
allow_model_passthrough=as_bool(
setting_value(
settings,
live_env,
"allow_model_passthrough",
"DEEPSEEK_ALLOW_MODEL_PASSTHROUGH",
),
False,
),
thinking=thinking,
reasoning_effort=values.get("DEEPSEEK_REASONING_EFFORT", "high"),
request_timeout=env_float(values, "PROXY_REQUEST_TIMEOUT", 300.0),
reasoning_content_path=env_path(
values,
("REASONING_CONTENT_PATH",),
default_reasoning_content_path(),
reasoning_effort=as_str(
setting_value(
settings,
live_env,
"reasoning_effort",
"DEEPSEEK_REASONING_EFFORT",
),
"high",
),
cursor_display_reasoning=env_bool(values, "CURSOR_DISPLAY_REASONING", True),
verbose=env_bool(values, "PROXY_VERBOSE", False),
log_bodies=env_bool(values, "PROXY_LOG_BODIES", False),
ngrok=env_bool(values, "PROXY_NGROK", False),
model_list=env_tuple(
values,
"PROXY_MODELS",
("deepseek-v4-pro", "deepseek-v4-flash"),
request_timeout=as_float(
setting_value(
settings,
live_env,
"request_timeout",
"PROXY_REQUEST_TIMEOUT",
),
300.0,
),
reasoning_content_path=as_path(
setting_value(
settings,
live_env,
"reasoning_content_path",
"REASONING_CONTENT_PATH",
),
default_reasoning_content_path(),
config_dir,
),
cursor_display_reasoning=as_bool(
setting_value(
settings,
live_env,
"display_reasoning",
"CURSOR_DISPLAY_REASONING",
),
True,
),
verbose=as_bool(
setting_value(
settings,
live_env,
"verbose",
"PROXY_VERBOSE",
),
False,
),
log_bodies=as_bool(
setting_value(
settings,
live_env,
"log_bodies",
"PROXY_LOG_BODIES",
),
False,
),
ngrok=as_bool(
setting_value(
settings,
live_env,
"ngrok",
"PROXY_NGROK",
),
False,
),
)
def validate(self) -> None:
if not self.upstream_api_key:
raise ValueError("DEEPSEEK_API_KEY is required")

View File

@ -88,18 +88,13 @@ class DeepSeekProxyHandler(BaseHTTPRequestHandler):
404, {"error": {"message": "Only /v1/chat/completions is supported"}}
)
return
if not self._authorized():
cursor_authorization = self._cursor_authorization()
if cursor_authorization is None:
self._send_json(
401, {"error": {"message": "Missing or invalid proxy API key"}}
401, {"error": {"message": "Missing Authorization bearer token"}}
)
return
try:
self.config.validate()
except ValueError as exc:
self._send_json(500, {"error": {"message": str(exc)}})
return
try:
payload = self._read_json_body()
except ValueError as exc:
@ -148,7 +143,10 @@ class DeepSeekProxyHandler(BaseHTTPRequestHandler):
upstream_url,
data=upstream_body,
method="POST",
headers=self._upstream_headers(stream=bool(prepared.payload.get("stream"))),
headers=self._upstream_headers(
stream=bool(prepared.payload.get("stream")),
authorization=cursor_authorization,
),
)
try:
@ -193,12 +191,12 @@ class DeepSeekProxyHandler(BaseHTTPRequestHandler):
response, prepared.original_model, prepared.payload["messages"]
)
def _authorized(self) -> bool:
expected = self.config.proxy_api_key
if expected is None:
return True
def _cursor_authorization(self) -> str | None:
auth_header = self.headers.get("Authorization", "")
return auth_header == f"Bearer {expected}"
scheme, separator, token = auth_header.strip().partition(" ")
if separator != " " or scheme.lower() != "bearer" or not token.strip():
return None
return f"Bearer {token.strip()}"
def _send_cors_headers(self) -> None:
self.send_header("Access-Control-Allow-Origin", "*")
@ -223,20 +221,14 @@ class DeepSeekProxyHandler(BaseHTTPRequestHandler):
def _send_models(self) -> None:
created = int(time.time())
seen: set[str] = set()
models = []
for model_id in (self.config.upstream_model, *self.config.model_list):
if model_id in seen:
continue
seen.add(model_id)
models.append(
{
"id": model_id,
"object": "model",
"created": created,
"owned_by": "deepseek",
}
)
models = [
{
"id": self.config.upstream_model,
"object": "model",
"created": created,
"owned_by": "deepseek",
}
]
self._send_json(200, {"object": "list", "data": models})
def _read_json_body(self) -> dict[str, Any]:
@ -252,9 +244,9 @@ class DeepSeekProxyHandler(BaseHTTPRequestHandler):
raise ValueError("Request body must be a JSON object")
return payload
def _upstream_headers(self, stream: bool) -> dict[str, str]:
def _upstream_headers(self, stream: bool, authorization: str) -> dict[str, str]:
headers = {
"Authorization": f"Bearer {self.config.upstream_api_key}",
"Authorization": authorization,
"Content-Type": "application/json",
"Accept": "text/event-stream" if stream else "application/json",
"Accept-Encoding": "identity",
@ -399,19 +391,23 @@ def build_arg_parser() -> argparse.ArgumentParser:
"--config",
dest="config_path",
type=Path,
help=f"Env config file, default {default_config_path()}",
help=f"YAML config file, default {default_config_path()}",
)
parser.add_argument(
"--host", help="Bind host, default from PROXY_HOST or 127.0.0.1"
"--host", help="Bind host, default from config, PROXY_HOST, or 127.0.0.1"
)
parser.add_argument(
"--port", type=int, help="Bind port, default from PROXY_PORT or 9000"
"--port",
type=int,
help="Bind port, default from config, PROXY_PORT, or 9000",
)
parser.add_argument(
"--model", help="Upstream DeepSeek model, default from DEEPSEEK_MODEL"
"--model",
help="Upstream DeepSeek model, default from config, DEEPSEEK_MODEL, or deepseek-v4-pro",
)
parser.add_argument(
"--base-url", help="DeepSeek base URL, default https://api.deepseek.com"
"--base-url",
help="DeepSeek base URL, default from config, DEEPSEEK_BASE_URL, or https://api.deepseek.com",
)
parser.add_argument(
"--reasoning-content-path",
@ -505,7 +501,11 @@ def main(argv: list[str] | None = None) -> int:
level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
)
args = build_arg_parser().parse_args(argv)
config = ProxyConfig.from_env(env_file_path=args.config_path)
try:
config = ProxyConfig.from_file(config_path=args.config_path)
except ValueError as exc:
LOG.error("%s", exc)
return 2
updates: dict[str, Any] = {}
if args.host:
updates["host"] = args.host
@ -528,12 +528,6 @@ def main(argv: list[str] | None = None) -> int:
if updates:
config = replace(config, **updates)
try:
config.validate()
except ValueError as exc:
LOG.error("%s", exc)
return 2
store = ReasoningStore(config.reasoning_content_path)
server = DeepSeekProxyServer((config.host, config.port), DeepSeekProxyHandler)
server.config = config
@ -572,8 +566,6 @@ def main(argv: list[str] | None = None) -> int:
return 2
LOG.info("ngrok tunnel forwarding %s -> %s", public_url, target_url)
LOG.info("Cursor Base URL: %s/v1", public_url.rstrip("/"))
if config.proxy_api_key:
LOG.info("Cursor API key: value of PROXY_API_KEY")
try:
server.serve_forever()
except KeyboardInterrupt:

View File

@ -2,6 +2,7 @@ from __future__ import annotations
import os
from pathlib import Path
import stat
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch
@ -14,12 +15,12 @@ from deepseek_cursor_proxy.config import (
class ConfigTests(unittest.TestCase):
def test_default_paths_live_in_user_app_directory(self) -> None:
def test_default_paths_live_in_visible_user_app_directory(self) -> None:
home = Path("/tmp/home")
with patch("deepseek_cursor_proxy.config.Path.home", return_value=home):
self.assertEqual(
default_config_path(), home / ".deepseek-cursor-proxy" / ".env"
default_config_path(), home / ".deepseek-cursor-proxy" / "config.yaml"
)
self.assertEqual(
default_reasoning_content_path(),
@ -30,63 +31,103 @@ class ConfigTests(unittest.TestCase):
home / ".deepseek-cursor-proxy" / "reasoning_content.sqlite3",
)
def test_loads_config_from_user_env_file(self) -> None:
def test_missing_default_config_file_is_populated(self) -> None:
with TemporaryDirectory() as temp_dir:
env_file_path = Path(temp_dir) / ".env"
home = Path(temp_dir)
with patch("deepseek_cursor_proxy.config.Path.home", return_value=home):
config = ProxyConfig.from_file(env={}, config_path=None)
config_path = default_config_path()
self.assertTrue(config_path.exists())
self.assertIn(
"model: deepseek-v4-pro", config_path.read_text(encoding="utf-8")
)
self.assertEqual(stat.S_IMODE(config_path.stat().st_mode), 0o600)
self.assertEqual(config.upstream_model, "deepseek-v4-pro")
def test_missing_explicit_config_file_is_not_populated(self) -> None:
with TemporaryDirectory() as temp_dir:
config_path = Path(temp_dir) / "missing.yaml"
config = ProxyConfig.from_file(env={}, config_path=config_path)
self.assertFalse(config_path.exists())
self.assertEqual(config.upstream_model, "deepseek-v4-pro")
def test_loads_config_from_user_yaml_file(self) -> None:
with TemporaryDirectory() as temp_dir:
config_path = Path(temp_dir) / "config.yaml"
reasoning_content_path = Path(temp_dir) / "reasoning_content.sqlite3"
env_file_path.write_text(
config_path.write_text(
"\n".join(
[
"DEEPSEEK_API_KEY=file-key",
"PROXY_API_KEY=cursor-token",
"PROXY_PORT=9100",
f"REASONING_CONTENT_PATH={reasoning_content_path}",
"model: deepseek-v4-flash",
"port: 9100",
f"reasoning_content_path: {reasoning_content_path}",
]
),
encoding="utf-8",
)
config = ProxyConfig.from_env(env={}, env_file_path=env_file_path)
config = ProxyConfig.from_file(env={}, config_path=config_path)
self.assertEqual(config.upstream_api_key, "file-key")
self.assertEqual(config.proxy_api_key, "cursor-token")
self.assertEqual(config.upstream_model, "deepseek-v4-flash")
self.assertEqual(config.port, 9100)
self.assertEqual(config.reasoning_content_path, reasoning_content_path)
def test_environment_overrides_config_file(self) -> None:
with TemporaryDirectory() as temp_dir:
env_file_path = Path(temp_dir) / ".env"
env_file_path.write_text(
config_path = Path(temp_dir) / "config.yaml"
config_path.write_text(
"\n".join(
[
"DEEPSEEK_API_KEY=file-key",
"PROXY_VERBOSE=false",
"verbose: false",
]
),
encoding="utf-8",
)
config = ProxyConfig.from_env(
config = ProxyConfig.from_file(
env={
"DEEPSEEK_API_KEY": "env-key",
"PROXY_VERBOSE": "true",
},
env_file_path=env_file_path,
config_path=config_path,
)
self.assertEqual(config.upstream_api_key, "env-key")
self.assertTrue(config.verbose)
def test_relative_reasoning_content_path_stays_inside_app_directory(self) -> None:
def test_relative_reasoning_content_path_in_config_is_relative_to_config_file(
self,
) -> None:
with TemporaryDirectory() as temp_dir:
config_path = Path(temp_dir) / "config.yaml"
config_path.write_text(
"\n".join(
[
"reasoning_content_path: custom.sqlite3",
]
),
encoding="utf-8",
)
config = ProxyConfig.from_file(env={}, config_path=config_path)
self.assertEqual(
config.reasoning_content_path, Path(temp_dir) / "custom.sqlite3"
)
def test_relative_reasoning_content_path_from_env_stays_inside_app_directory(
self,
) -> None:
home = Path("/tmp/home")
with patch("deepseek_cursor_proxy.config.Path.home", return_value=home):
config = ProxyConfig.from_env(
config = ProxyConfig.from_file(
env={
"DEEPSEEK_API_KEY": "key",
"REASONING_CONTENT_PATH": "custom.sqlite3",
},
env_file_path=Path("/does/not/exist"),
config_path=None,
)
self.assertEqual(
@ -95,71 +136,83 @@ class ConfigTests(unittest.TestCase):
)
def test_verbose_and_body_logging_can_be_enabled_from_env(self) -> None:
config = ProxyConfig.from_env(
config = ProxyConfig.from_file(
env={
"DEEPSEEK_API_KEY": "key",
"PROXY_VERBOSE": "true",
"PROXY_LOG_BODIES": "1",
"PROXY_NGROK": "yes",
},
env_file_path=Path("/does/not/exist"),
config_path=Path("/does/not/exist"),
)
self.assertTrue(config.verbose)
self.assertTrue(config.log_bodies)
self.assertTrue(config.ngrok)
def test_cursor_reasoning_display_can_be_disabled_from_env(self) -> None:
config = ProxyConfig.from_env(
env={
"DEEPSEEK_API_KEY": "key",
"CURSOR_DISPLAY_REASONING": "false",
},
env_file_path=Path("/does/not/exist"),
)
def test_cursor_reasoning_display_can_be_disabled_from_config(self) -> None:
with TemporaryDirectory() as temp_dir:
config_path = Path(temp_dir) / "config.yaml"
config_path.write_text(
"\n".join(
[
"display_reasoning: false",
]
),
encoding="utf-8",
)
config = ProxyConfig.from_file(env={}, config_path=config_path)
self.assertFalse(config.cursor_display_reasoning)
def test_config_path_can_be_overridden_from_environment(self) -> None:
with TemporaryDirectory() as temp_dir:
first_env_path = Path(temp_dir) / "first.env"
second_env_path = Path(temp_dir) / "second.env"
first_env_path.write_text("DEEPSEEK_API_KEY=first-key", encoding="utf-8")
second_env_path.write_text("DEEPSEEK_API_KEY=second-key", encoding="utf-8")
first_config_path = Path(temp_dir) / "first.yaml"
second_config_path = Path(temp_dir) / "second.yaml"
first_config_path.write_text("port: 9100\n", encoding="utf-8")
second_config_path.write_text("port: 9200\n", encoding="utf-8")
config = ProxyConfig.from_env(
env={"DEEPSEEK_CURSOR_PROXY_CONFIG_PATH": str(second_env_path)},
env_file_path=None,
config = ProxyConfig.from_file(
env={"DEEPSEEK_CURSOR_PROXY_CONFIG_PATH": str(second_config_path)},
config_path=None,
)
self.assertEqual(config.upstream_api_key, "second-key")
self.assertEqual(config.port, 9200)
def test_explicit_env_file_path_wins_over_config_path_environment_variable(
def test_explicit_config_file_path_wins_over_config_path_environment_variable(
self,
) -> None:
with TemporaryDirectory() as temp_dir:
first_env_path = Path(temp_dir) / "first.env"
second_env_path = Path(temp_dir) / "second.env"
first_env_path.write_text("DEEPSEEK_API_KEY=first-key", encoding="utf-8")
second_env_path.write_text("DEEPSEEK_API_KEY=second-key", encoding="utf-8")
first_config_path = Path(temp_dir) / "first.yaml"
second_config_path = Path(temp_dir) / "second.yaml"
first_config_path.write_text("port: 9100\n", encoding="utf-8")
second_config_path.write_text("port: 9200\n", encoding="utf-8")
config = ProxyConfig.from_env(
env={"DEEPSEEK_CURSOR_PROXY_CONFIG_PATH": str(second_env_path)},
env_file_path=first_env_path,
config = ProxyConfig.from_file(
env={"DEEPSEEK_CURSOR_PROXY_CONFIG_PATH": str(second_config_path)},
config_path=first_config_path,
)
self.assertEqual(config.upstream_api_key, "first-key")
self.assertEqual(config.port, 9100)
def test_from_env_does_not_mutate_process_environment(self) -> None:
def test_invalid_yaml_config_raises_value_error(self) -> None:
with TemporaryDirectory() as temp_dir:
config_path = Path(temp_dir) / "config.yaml"
config_path.write_text("- not\n- a\n- mapping\n", encoding="utf-8")
with self.assertRaises(ValueError):
ProxyConfig.from_file(env={}, config_path=config_path)
def test_from_file_does_not_mutate_process_environment(self) -> None:
with patch.dict(
"os.environ",
{
"DEEPSEEK_API_KEY": "env-key",
"PROXY_VERBOSE": "true",
},
clear=True,
):
ProxyConfig.from_env(env_file_path=Path("/does/not/exist"))
self.assertEqual(dict(os.environ), {"DEEPSEEK_API_KEY": "env-key"})
ProxyConfig.from_file(config_path=Path("/does/not/exist"))
self.assertEqual(dict(os.environ), {"PROXY_VERBOSE": "true"})
if __name__ == "__main__":

View File

@ -13,7 +13,9 @@ from deepseek_cursor_proxy.reasoning_store import ReasoningStore
from deepseek_cursor_proxy.server import DeepSeekProxyHandler, DeepSeekProxyServer
LIVE_DEEPSEEK = os.getenv("RUN_LIVE_DEEPSEEK_TESTS") == "1"
LIVE_DEEPSEEK = os.getenv("RUN_LIVE_DEEPSEEK_TESTS") == "1" and bool(
os.getenv("LIVE_DEEPSEEK_KEY")
)
def post_json(
@ -37,12 +39,10 @@ def post_json(
class ProxyFixture:
def __init__(self, api_key: str) -> None:
def __init__(self) -> None:
self.store = ReasoningStore(":memory:")
server = DeepSeekProxyServer(("127.0.0.1", 0), DeepSeekProxyHandler)
server.config = ProxyConfig(
upstream_api_key=api_key,
proxy_api_key="cursor-local-token",
upstream_base_url="https://api.deepseek.com",
upstream_model="deepseek-v4-pro",
request_timeout=180,
@ -68,17 +68,18 @@ class ProxyFixture:
@unittest.skipUnless(
LIVE_DEEPSEEK, "set RUN_LIVE_DEEPSEEK_TESTS=1 to run live DeepSeek API tests"
LIVE_DEEPSEEK,
"set RUN_LIVE_DEEPSEEK_TESTS=1 and LIVE_DEEPSEEK_KEY to run live tests",
)
class LiveDeepSeekProxyTests(unittest.TestCase):
def test_proxy_repairs_real_deepseek_tool_call_history(self) -> None:
api_key = os.environ["DEEPSEEK_API_KEY"]
proxy = ProxyFixture(api_key).start()
api_key = os.environ["LIVE_DEEPSEEK_KEY"]
proxy = ProxyFixture().start()
try:
first_status, first_response = post_json(
proxy.url,
first_request(),
api_key="cursor-local-token",
api_key=api_key,
)
self.assertEqual(first_status, 200, first_response.get("error"))
assistant_with_reasoning = first_response["choices"][0]["message"]
@ -118,7 +119,7 @@ class LiveDeepSeekProxyTests(unittest.TestCase):
proxy_status, second_response = post_json(
proxy.url,
missing_reasoning_payload,
api_key="cursor-local-token",
api_key=api_key,
)
self.assertEqual(proxy_status, 200, second_response.get("error"))
final_assistant = second_response["choices"][0]["message"]
@ -145,7 +146,7 @@ class LiveDeepSeekProxyTests(unittest.TestCase):
followup_status, followup_response = post_json(
proxy.url,
followup_payload,
api_key="cursor-local-token",
api_key=api_key,
)
self.assertEqual(followup_status, 200, followup_response.get("error"))
finally:

View File

@ -23,7 +23,7 @@ FINAL_CONTENT = "Final answer after using the tool."
def post_json(
url: str, payload: dict, api_key: str = "cursor-local-token"
url: str, payload: dict, api_key: str = "sk-cursor-test"
) -> tuple[int, dict]:
body = json.dumps(payload).encode("utf-8")
request = Request(
@ -45,6 +45,7 @@ def post_json(
class FakeDeepSeekHandler(BaseHTTPRequestHandler):
requests: list[dict] = []
auth_headers: list[str] = []
def log_message(self, fmt: str, *args: object) -> None:
return
@ -53,6 +54,7 @@ class FakeDeepSeekHandler(BaseHTTPRequestHandler):
length = int(self.headers.get("Content-Length") or 0)
payload = json.loads(self.rfile.read(length).decode("utf-8"))
self.__class__.requests.append(payload)
self.__class__.auth_headers.append(self.headers.get("Authorization", ""))
for index, message in enumerate(payload.get("messages", [])):
if not isinstance(message, dict) or message.get("role") != "assistant":
@ -383,14 +385,13 @@ class ServerFixture:
class ProxyEndToEndTests(unittest.TestCase):
def setUp(self) -> None:
FakeDeepSeekHandler.requests = []
FakeDeepSeekHandler.auth_headers = []
self.upstream = ServerFixture(
ThreadingHTTPServer(("127.0.0.1", 0), FakeDeepSeekHandler)
).start()
self.store = ReasoningStore(":memory:")
proxy = DeepSeekProxyServer(("127.0.0.1", 0), DeepSeekProxyHandler)
proxy.config = ProxyConfig(
upstream_api_key="upstream-key",
proxy_api_key="cursor-local-token",
upstream_base_url=self.upstream.url,
upstream_model="deepseek-v4-pro",
)
@ -451,6 +452,30 @@ class ProxyEndToEndTests(unittest.TestCase):
third_upstream_messages[3]["reasoning_content"], FINAL_REASONING
)
def test_proxy_forwards_cursor_bearer_token_to_deepseek(self) -> None:
status, _ = post_json(
f"{self.proxy.url}/v1/chat/completions",
first_cursor_request(),
api_key="sk-from-cursor",
)
self.assertEqual(status, 200)
self.assertEqual(FakeDeepSeekHandler.auth_headers[0], "Bearer sk-from-cursor")
def test_proxy_rejects_missing_cursor_bearer_token(self) -> None:
request = Request(
f"{self.proxy.url}/v1/chat/completions",
data=json.dumps(first_cursor_request()).encode("utf-8"),
method="POST",
headers={"Content-Type": "application/json"},
)
with self.assertRaises(HTTPError) as caught:
urlopen(request, timeout=5)
self.assertEqual(caught.exception.code, 401)
self.assertEqual(FakeDeepSeekHandler.requests, [])
def test_proxy_adds_fallback_reasoning_for_uncached_cursor_tool_history(
self,
) -> None:
@ -473,8 +498,6 @@ class InterleavedConversationTests(unittest.TestCase):
self.store = ReasoningStore(":memory:")
proxy = DeepSeekProxyServer(("127.0.0.1", 0), DeepSeekProxyHandler)
proxy.config = ProxyConfig(
upstream_api_key="upstream-key",
proxy_api_key="cursor-local-token",
upstream_base_url=self.upstream.url,
upstream_model="deepseek-v4-pro",
)
@ -580,8 +603,6 @@ class StreamingProxyTests(unittest.TestCase):
self.store = ReasoningStore(":memory:")
proxy = DeepSeekProxyServer(("127.0.0.1", 0), DeepSeekProxyHandler)
proxy.config = ProxyConfig(
upstream_api_key="upstream-key",
proxy_api_key="cursor-local-token",
upstream_base_url=self.upstream.url,
upstream_model="deepseek-v4-pro",
)
@ -607,7 +628,7 @@ class StreamingProxyTests(unittest.TestCase):
).encode("utf-8"),
method="POST",
headers={
"Authorization": "Bearer cursor-local-token",
"Authorization": "Bearer sk-cursor-test",
"Content-Type": "application/json",
},
)
@ -629,8 +650,6 @@ class ReasoningStreamingProxyTests(unittest.TestCase):
self.store = ReasoningStore(":memory:")
proxy = DeepSeekProxyServer(("127.0.0.1", 0), DeepSeekProxyHandler)
proxy.config = ProxyConfig(
upstream_api_key="upstream-key",
proxy_api_key="cursor-local-token",
upstream_base_url=self.upstream.url,
upstream_model="deepseek-v4-pro",
)
@ -657,7 +676,7 @@ class ReasoningStreamingProxyTests(unittest.TestCase):
).encode("utf-8"),
method="POST",
headers={
"Authorization": "Bearer cursor-local-token",
"Authorization": "Bearer sk-cursor-test",
"Content-Type": "application/json",
},
)

View File

@ -57,9 +57,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.payload["messages"][1]["content"], "Visible answer.")
@ -72,7 +70,7 @@ class TransformTests(unittest.TestCase):
"max_completion_tokens": 123,
"parallel_tool_calls": True,
}
config = ProxyConfig(upstream_api_key="key")
config = ProxyConfig()
prepared = prepare_upstream_request(payload, config, self.store)
@ -97,9 +95,7 @@ class TransformTests(unittest.TestCase):
"tool_choice": "required",
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.payload["tool_choice"], "auto")
@ -147,9 +143,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.patched_reasoning_messages, 1)
self.assertEqual(
@ -212,9 +206,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.patched_reasoning_messages, 1)
self.assertEqual(
@ -261,12 +253,8 @@ class TransformTests(unittest.TestCase):
],
}
prepared_a = prepare_upstream_request(
payload_a, ProxyConfig(upstream_api_key="key"), self.store
)
prepared_b = prepare_upstream_request(
payload_b, ProxyConfig(upstream_api_key="key"), self.store
)
prepared_a = prepare_upstream_request(payload_a, ProxyConfig(), self.store)
prepared_b = prepare_upstream_request(payload_b, ProxyConfig(), self.store)
self.assertEqual(
prepared_a.payload["messages"][1]["reasoning_content"],
@ -321,7 +309,7 @@ class TransformTests(unittest.TestCase):
},
],
},
ProxyConfig(upstream_api_key="key"),
ProxyConfig(),
self.store,
)
@ -370,9 +358,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.patched_reasoning_messages, 1)
self.assertEqual(prepared.payload["messages"][1]["content"], "")
@ -415,7 +401,7 @@ class TransformTests(unittest.TestCase):
},
],
},
ProxyConfig(upstream_api_key="key"),
ProxyConfig(),
self.store,
)
@ -453,9 +439,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.patched_reasoning_messages, 0)
self.assertEqual(prepared.fallback_reasoning_messages, 1)
@ -493,9 +477,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.fallback_reasoning_messages, 1)
self.assertIn("reasoning_content", prepared.payload["messages"][3])
@ -510,9 +492,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(prepared.fallback_reasoning_messages, 0)
self.assertNotIn("reasoning_content", prepared.payload["messages"][1])
@ -530,9 +510,7 @@ class TransformTests(unittest.TestCase):
],
}
prepared = prepare_upstream_request(
payload, ProxyConfig(upstream_api_key="key"), self.store
)
prepared = prepare_upstream_request(payload, ProxyConfig(), self.store)
self.assertEqual(
prepared.payload["messages"][0],