Files
zk-data-agent/agent_platform/gateway/schemas.py
T
2026-07-26 15:10:56 +08:00

105 lines
3.0 KiB
Python

from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field, field_validator
class PathRequest(BaseModel):
path: str = Field(default=".", max_length=4096, description="Path relative to /workspace.")
class ListFilesRequest(PathRequest):
max_depth: int = Field(default=4, ge=1, le=12)
limit: int = Field(default=500, ge=1, le=5000)
class ReadFileRequest(PathRequest):
start_line: int = Field(default=1, ge=1)
max_lines: int = Field(default=1000, ge=1, le=5000)
class WriteFileRequest(PathRequest):
content: str = Field(max_length=2_000_000, description="Complete UTF-8 file content.")
class SearchFilesRequest(BaseModel):
query: str = Field(min_length=1, max_length=500)
path: str = Field(default=".", description="Directory relative to /workspace.")
glob: str | None = Field(default=None, max_length=200)
limit: int = Field(default=200, ge=1, le=2000)
class ExecRequest(BaseModel):
command: str = Field(min_length=1, max_length=32_000)
cwd: str = Field(default=".", max_length=4096, description="Working directory relative to /workspace.")
timeout_seconds: int = Field(default=120, ge=1, le=1800)
class ApplyPatchRequest(BaseModel):
patch: str = Field(min_length=1, max_length=512_000)
cwd: str = Field(default=".", max_length=4096, description="Repository directory relative to /workspace.")
class GitRequest(BaseModel):
cwd: str = Field(default=".", max_length=4096, description="Repository directory relative to /workspace.")
class GitDiffRequest(GitRequest):
staged: bool = False
class StartProcessRequest(BaseModel):
command: str = Field(min_length=1, max_length=32_000)
cwd: str = Field(default=".", max_length=4096, description="Working directory relative to /workspace.")
class ProcessRequest(BaseModel):
process_id: str = Field(pattern=r"^[a-f0-9]{32}$")
class ToolResult(BaseModel):
ok: bool
output: str = ""
exit_code: int | None = None
truncated: bool = False
metadata: dict = Field(default_factory=dict)
class WorkspaceStatus(BaseModel):
ok: bool = True
workspace_id: str
provider: Literal["local-docker", "ssh-docker"]
container_name: str
state: str
class WorkspaceEntry(BaseModel):
name: str
path: str
kind: Literal["file", "directory", "symlink"]
size: int = Field(ge=0)
modified_at: int = Field(ge=0)
class WorkspaceListing(BaseModel):
path: str
entries: list[WorkspaceEntry]
class PlanItem(BaseModel):
step: str = Field(min_length=1, max_length=1000)
status: Literal["pending", "in_progress", "completed"]
class UpdatePlanRequest(BaseModel):
explanation: str | None = Field(default=None, max_length=4000)
items: list[PlanItem] = Field(min_length=1, max_length=50)
@field_validator("items")
@classmethod
def one_in_progress(cls, value: list[PlanItem]) -> list[PlanItem]:
if sum(item.status == "in_progress" for item in value) > 1:
raise ValueError("At most one plan item may be in progress")
return value