schemas.py•5.44 kB
"""
Pydantic models for request/response schemas.
Defines the data structures for the MCP debug tool API.
"""
from enum import Enum
from typing import Any
from pydantic import BaseModel, Field, field_validator
class SessionStatus(str, Enum):
"""Debug session status."""
IDLE = "idle"
RUNNING = "running"
PAUSED = "paused"
COMPLETED = "completed"
ERROR = "error"
class StartSessionRequest(BaseModel):
"""Request to start a new debug session."""
entry: str = Field(..., description="Project-relative path to script")
pythonPath: str = Field(..., description="Absolute path to Python interpreter (must have debugpy installed)")
args: list[str] | None = Field(default=None, description="Command-line arguments")
env: dict[str, str] | None = Field(default=None, description="Environment variables")
useDap: bool | None = Field(default=True, description="Use DAP (debugpy) instead of bdb - recommended for all use cases")
@field_validator('args')
@classmethod
def validate_args(cls, v: list[str] | None) -> list[str] | None:
"""Validate args list constraints."""
if v is not None:
from .utils import MAX_ARG_LENGTH, MAX_ARGS_COUNT
if len(v) > MAX_ARGS_COUNT:
raise ValueError(f"Too many args (max {MAX_ARGS_COUNT})")
for arg in v:
if len(arg) > MAX_ARG_LENGTH:
raise ValueError(f"Arg too long (max {MAX_ARG_LENGTH} chars)")
return v
@field_validator('env')
@classmethod
def validate_env(cls, v: dict[str, str] | None) -> dict[str, str] | None:
"""Validate env map constraints."""
if v is not None:
from .utils import MAX_ENV_ENTRIES, MAX_ENV_KEY_LENGTH, MAX_ENV_VALUE_LENGTH
if len(v) > MAX_ENV_ENTRIES:
raise ValueError(f"Too many env vars (max {MAX_ENV_ENTRIES})")
for key, val in v.items():
if len(key) > MAX_ENV_KEY_LENGTH:
raise ValueError(f"Env key too long (max {MAX_ENV_KEY_LENGTH} chars)")
if len(val) > MAX_ENV_VALUE_LENGTH:
raise ValueError(f"Env value too long (max {MAX_ENV_VALUE_LENGTH} chars)")
return v
class StartSessionResponse(BaseModel):
"""Response with new session ID."""
sessionId: str
class BreakpointRequest(BaseModel):
"""Request to run to a breakpoint."""
file: str = Field(..., description="Project-relative path to file")
line: int = Field(..., ge=1, description="1-based line number")
class ContinueRequest(BaseModel):
"""Request to continue to next breakpoint."""
file: str = Field(..., description="Project-relative path to file")
line: int = Field(..., ge=1, description="1-based line number")
class StepRequest(BaseModel):
"""Request for step operation (no parameters needed - uses current thread state)."""
pass
class FrameInfo(BaseModel):
"""Stack frame information."""
file: str
line: int
class ExecutionError(BaseModel):
"""Structured error information."""
type: str = Field(..., description="Error type (e.g., 'SyntaxError', 'TimeoutError')")
message: str = Field(..., description="Error message")
traceback: str | None = Field(default=None, description="Truncated traceback")
class VariableValue(BaseModel):
"""Representation of a variable value."""
name: str
type: str
repr: str
size: int | None = None # Length for collections
isTruncated: bool = False
children: list['VariableValue'] | None = None
# Enable forward references
VariableValue.model_rebuild()
class BreakpointResponse(BaseModel):
"""
Response from run_to_breakpoint.
The locals field contains a dictionary where:
- Key: variable name (str)
- Value: dict with the following fields:
- type: Python type name (str)
- repr: string representation of the value (str)
- isTruncated: whether the repr was truncated (bool)
- length: for indexed collections like list/tuple (int, optional)
- isIndexed: true if this is an indexed collection (bool, optional)
- namedCount: number of named properties for dict/objects (int, optional)
- hasNamedProperties: true if has named properties (bool, optional)
- isExpandable: true if variable has children (bool, optional)
- variablesReference: DAP reference for fetching children (int, optional)
"""
hit: bool
frameInfo: FrameInfo | None = None
locals: dict[str, Any] | None = None
completed: bool = False
error: ExecutionError | None = None
class ContinueResponse(BaseModel):
"""Response from continue."""
hit: bool
completed: bool
frameInfo: FrameInfo | None = None
locals: dict[str, Any] | None = None
error: ExecutionError | None = None
class BreakpointTarget(BaseModel):
"""A breakpoint location."""
file: str
line: int
hitCount: int = 0
class SessionTimings(BaseModel):
"""Timing information for a session."""
lastRunMs: float | None = None
totalCpuTimeMs: float = 0.0
class SessionStateResponse(BaseModel):
"""Current session state."""
status: SessionStatus
lastBreakpoint: BreakpointTarget | None = None
timings: SessionTimings | None = None
class EndSessionResponse(BaseModel):
"""Response confirming session ended."""
ended: bool = True