diff --git a/app/api/proxy/router.py b/app/api/proxy/router.py
index 66481ca..0f662be 100644
--- a/app/api/proxy/router.py
+++ b/app/api/proxy/router.py
@@ -10,6 +10,7 @@ from db.proxy import (
 )
 from fastapi import APIRouter, Depends, HTTPException, status
 from models.proxy import ProxyCreateInput, ProxyUpdateInput
+from utils.proxy.validator import ProxyCodeValidationError, validate_proxy_code
 
 router = APIRouter(
     prefix="/proxy",
@@ -18,6 +19,16 @@ router = APIRouter(
 )
 
 
+async def _validate_code_or_raise(code: str) -> None:
+    try:
+        await validate_proxy_code(code)
+    except ProxyCodeValidationError as error:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=error.to_http_detail(),
+        ) from error
+
+
 @router.get("")
 async def list_proxies_handler():
     return await list_proxy_documents()
@@ -43,6 +54,8 @@ async def create_proxy_handler(payload: ProxyCreateInput):
             detail=f"Proxy already exists on port {payload.proxy_port}",
         )
 
+    await _validate_code_or_raise(payload.code)
+
     return await create_proxy(
         proxy_port=payload.proxy_port,
         name=payload.name,
@@ -53,6 +66,9 @@ async def create_proxy_handler(payload: ProxyCreateInput):
 
 @router.put("/{proxy_port}")
 async def edit_proxy(proxy_port: int, payload: ProxyUpdateInput):
+    if payload.code is not None:
+        await _validate_code_or_raise(payload.code)
+
     updated_proxy = await update_proxy(proxy_port, payload)
     if updated_proxy is None:
         raise HTTPException(
diff --git a/app/utils/proxy/validator.py b/app/utils/proxy/validator.py
new file mode 100644
index 0000000..69ae67a
--- /dev/null
+++ b/app/utils/proxy/validator.py
@@ -0,0 +1,164 @@
+import asyncio
+import json
+import os
+import shutil
+import tempfile
+from typing import Any
+
+TARGET_RUFF_CODES = ("F821", "F822", "F823", "E999")
+
+
+class ProxyCodeValidationError(Exception):
+    def __init__(
+        self,
+        message: str,
+        *,
+        errors: list[dict[str, Any]] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.errors = errors or []
+
+    def to_http_detail(self) -> dict[str, Any]:
+        return {
+            "message": self.message,
+            "errors": self.errors,
+        }
+
+
+def _validate_python_syntax(code: str) -> None:
+    try:
+        compile(code, "<proxy_code>", "exec")
+    except SyntaxError as error:
+        line = error.lineno if error.lineno is not None else 0
+        column = error.offset if error.offset is not None else 0
+        detail = {
+            "code": "COMPILE",
+            "message": error.msg,
+            "location": {"row": line, "column": column},
+        }
+        raise ProxyCodeValidationError(
+            "Proxy code has invalid Python syntax",
+            errors=[detail],
+        ) from error
+
+
+def _get_ruff_executable() -> list[str]:
+    ruff_path = shutil.which("ruff")
+    if ruff_path:
+        return [ruff_path]
+
+    uv_path = shutil.which("uv")
+    if uv_path:
+        return [uv_path, "run", "ruff"]
+
+    raise ProxyCodeValidationError("Unable to run ruff: executable not found")
+
+
+def _parse_ruff_output(output: str) -> list[dict[str, Any]]:
+    if not output.strip():
+        return []
+
+    try:
+        parsed = json.loads(output)
+    except json.JSONDecodeError as error:
+        raise ProxyCodeValidationError("Unable to parse ruff JSON output") from error
+
+    if not isinstance(parsed, list):
+        raise ProxyCodeValidationError("Unexpected ruff output format")
+
+    errors: list[dict[str, Any]] = []
+    for item in parsed:
+        if not isinstance(item, dict):
+            continue
+        code = item.get("code")
+        if code not in TARGET_RUFF_CODES:
+            continue
+
+        message = item.get("message")
+        raw_location = item.get("location")
+        location = raw_location if isinstance(raw_location, dict) else {}
+        row = location.get("row", 0)
+        column = location.get("column", 0)
+
+        errors.append(
+            {
+                "code": code,
+                "message": str(message) if message is not None else "",
+                "location": {"row": row, "column": column},
+            }
+        )
+
+    return errors
+
+
+async def _run_ruff_check(code: str) -> list[dict[str, Any]]:
+    command = _get_ruff_executable()
+
+    file_path = ""
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            suffix=".py",
+            encoding="utf-8",
+            delete=False,
+        ) as handle:
+            handle.write(code)
+            file_path = handle.name
+
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            "check",
+            "--output-format=json",
+            "--select",
+            ",".join(TARGET_RUFF_CODES),
+            file_path,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+    finally:
+        if file_path:
+            try:
+                os.unlink(file_path)
+            except FileNotFoundError:
+                pass
+
+    stdout_text = stdout.decode("utf-8", errors="replace")
+    stderr_text = stderr.decode("utf-8", errors="replace")
+
+    if process.returncode not in (0, 1):
+        raise ProxyCodeValidationError(
+            "Ruff check failed to execute",
+            errors=[
+                {
+                    "code": "RUFF",
+                    "message": stderr_text.strip() or "Unknown ruff failure",
+                    "location": {"row": 0, "column": 0},
+                }
+            ],
+        )
+
+    if stderr_text.strip() and process.returncode == 0:
+        raise ProxyCodeValidationError(
+            "Ruff emitted unexpected stderr output",
+            errors=[
+                {
+                    "code": "RUFF",
+                    "message": stderr_text.strip(),
+                    "location": {"row": 0, "column": 0},
+                }
+            ],
+        )
+
+    return _parse_ruff_output(stdout_text)
+
+
+async def validate_proxy_code(code: str) -> None:
+    _validate_python_syntax(code)
+    errors = await _run_ruff_check(code)
+    if errors:
+        raise ProxyCodeValidationError(
+            "Proxy code contains ruff violations",
+            errors=errors,
+        )
diff --git a/tests/api/test_proxy.py b/tests/api/test_proxy.py
index d521a8f..27962f6 100644
--- a/tests/api/test_proxy.py
+++ b/tests/api/test_proxy.py
@@ -3,6 +3,7 @@ import importlib
 from core.security import get_auth_context, require_authenticated_user
 from models.proxy import ProxyDocument
 from models.service import ServiceEndpointDocument
+from utils.proxy.validator import ProxyCodeValidationError
 
 proxy_router_module = importlib.import_module("api.proxy.router")
 
@@ -120,11 +121,19 @@ def test_create_proxy_returns_created_document(client, monkeypatch: object):
             port=int(service["port"]),
         )
 
+    async def fake_validate_proxy_code(code: str) -> None:
+        assert code == "def proxy_handler():\n    return 'ok'\n"
+
     monkeypatch.setattr(
         proxy_router_module,
         "get_proxy_by_port",
         fake_get_proxy_by_port,
     )
+    monkeypatch.setattr(
+        proxy_router_module,
+        "validate_proxy_code",
+        fake_validate_proxy_code,
+    )
     monkeypatch.setattr(proxy_router_module, "create_proxy", fake_create_proxy)
 
     response = client.post(
@@ -132,7 +141,7 @@ def test_create_proxy_returns_created_document(client, monkeypatch: object):
         json={
             "proxy_port": 4040,
             "name": "edge",
-            "code": "edge-code",
+            "code": "def proxy_handler():\n    return 'ok'\n",
             "service": {"address": "service.local", "port": 8181},
         },
     )
@@ -143,5 +152,114 @@ def test_create_proxy_returns_created_document(client, monkeypatch: object):
         "proxy_port": 4040,
         "name": "edge",
         "service": {"address": "service.local", "port": 8181},
-        "code": "edge-code",
+        "code": "def proxy_handler():\n    return 'ok'\n",
     }
+
+
+def test_create_proxy_returns_422_when_code_validation_fails(
+    client,
+    monkeypatch: object,
+):
+    create_called = False
+
+    async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
+        assert proxy_port == 4000
+        return None
+
+    async def fake_validate_proxy_code(code: str) -> None:
+        assert code == "invalid-code"
+        raise ProxyCodeValidationError(
+            "Proxy code contains ruff violations",
+            errors=[
+                {
+                    "code": "F821",
+                    "message": "Undefined name `missing_symbol`",
+                    "location": {"row": 1, "column": 1},
+                }
+            ],
+        )
+
+    async def fake_create_proxy(
+        *,
+        proxy_port: int,
+        name: str,
+        service: dict[str, object],
+        code: str,
+    ) -> ProxyDocument:
+        nonlocal create_called
+        create_called = True
+        return make_proxy_document(proxy_port=proxy_port, name=name, code=code)
+
+    monkeypatch.setattr(
+        proxy_router_module,
+        "get_proxy_by_port",
+        fake_get_proxy_by_port,
+    )
+    monkeypatch.setattr(
+        proxy_router_module,
+        "validate_proxy_code",
+        fake_validate_proxy_code,
+    )
+    monkeypatch.setattr(proxy_router_module, "create_proxy", fake_create_proxy)
+
+    response = client.post(
+        "/proxy",
+        json={
+            "proxy_port": 4000,
+            "name": "broken",
+            "code": "invalid-code",
+            "service": {"address": "service.local", "port": 9000},
+        },
+    )
+
+    assert response.status_code == 422
+    body = response.json()["detail"]
+    assert body["message"] == "Proxy code contains ruff violations"
+    assert body["errors"][0]["code"] == "F821"
+    assert create_called is False
+
+
+def test_update_proxy_returns_422_when_code_validation_fails(
+    client,
+    monkeypatch: object,
+):
+    update_called = False
+
+    async def fake_validate_proxy_code(code: str) -> None:
+        assert code == "invalid-code"
+        raise ProxyCodeValidationError(
+            "Proxy code has invalid Python syntax",
+            errors=[
+                {
+                    "code": "COMPILE",
+                    "message": "invalid syntax",
+                    "location": {"row": 1, "column": 1},
+                }
+            ],
+        )
+
+    async def fake_update_proxy(
+        proxy_port: int,
+        payload: object,
+    ) -> ProxyDocument | None:
+        nonlocal update_called
+        update_called = True
+        return make_proxy_document(proxy_port=proxy_port)
+
+    monkeypatch.setattr(
+        proxy_router_module,
+        "validate_proxy_code",
+        fake_validate_proxy_code,
+    )
+    monkeypatch.setattr(proxy_router_module, "update_proxy", fake_update_proxy)
+
+    response = client.put(
+        "/proxy/8123",
+        json={"code": "invalid-code"},
+    )
+
+    assert response.status_code == 422
+    body = response.json()["detail"]
+    assert body["message"] == "Proxy code has invalid Python syntax"
+    assert body["errors"][0]["code"] == "COMPILE"
+    assert update_called is False