Schema Reference

Complete reference for warp-md’s Pydantic schemas and agent contract. All schemas are defined in python/warp_md/agent_schema.py.

Schema Version

AGENT_REQUEST_SCHEMA_VERSION = "warp-md.agent.v1"
AGENT_RESULT_SCHEMA_VERSION = "warp-md.agent.v1"

The schema version ensures backward compatibility. All requests and responses include a version or schema_version field.

Request Schema

RunRequest

The top-level request schema for batch analyses:

class RunRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    version: str = AGENT_REQUEST_SCHEMA_VERSION
    run_id: Optional[str] = None
    system: Optional[Dict[str, Any]] = None
    topology: Optional[Dict[str, Any]] = None
    trajectory: Optional[Dict[str, Any]] = None
    traj: Optional[Dict[str, Any]] = None
    device: str = "auto"
    stream: Literal["none", "ndjson"] = "none"
    chunk_frames: Optional[int] = Field(default=None, ge=1)
    output_dir: str = "."
    checkpoint: Optional[CheckpointConfig] = None
    fail_fast: bool = True
    analyses: list[AnalysisRequest] = Field(min_length=1)

Field coercion: String paths are automatically coerced to spec dictionaries. For example, "system": "protein.pdb" becomes "system": {"path": "protein.pdb"}.

Fields

Field	Type	Default	Description
`version`	`str`	`"warp-md.agent.v1"`	Schema version (required)
`run_id`	`str?`	`None`	Optional identifier for this run
`system`	`dict?`	`None`	System/topology file spec (use `system` OR `topology`)
`topology`	`dict?`	`None`	Alias for `system`
`trajectory`	`dict?`	`None`	Trajectory file spec (use `trajectory` OR `traj`)
`traj`	`dict?`	`None`	Alias for `trajectory`
`device`	`str`	`"auto"`	Compute device: `"auto"`, `"cpu"`, `"cuda"`, `"cuda:0"`
`stream`	`str`	`"none"`	Streaming mode: `"none"` or `"ndjson"`
`chunk_frames`	`int?`	`None`	Frames per chunk (memory control)
`output_dir`	`str`	`"."`	Output directory for results
`checkpoint`	`CheckpointConfig?`	`None`	Checkpoint event configuration
`fail_fast`	`bool`	`True`	Stop on first error
`analyses`	`list[AnalysisRequest]`	Required	List of analyses to run

Validation Rules

Must specify exactly one of system or topology
Must specify exactly one of trajectory or traj
version must match AGENT_REQUEST_SCHEMA_VERSION
analyses must have at least one entry
No extra fields allowed (extra="forbid")

AnalysisRequest

Specification for a single analysis:

class AnalysisRequest(BaseModel):
    model_config = ConfigDict(extra="allow")

    name: AnalysisName
    out: Optional[str] = None
    device: Optional[str] = None
    chunk_frames: Optional[int] = Field(default=None, ge=1)

Analysis-specific parameters (e.g., selection, bins, r_max) are stored in __pydantic_extra__ and validated against analysis requirements.

Common Analysis Parameters

Analysis	Required Fields	Optional Fields
`rg`	`selection`	`out`, `device`, `chunk_frames`
`rmsd`	`selection`	`align`, `ref_frame`, `out`
`msd`	`selection`	`out`, `device`
`rdf`	`sel_a`, `sel_b`, `bins`, `r_max`	`out`, `device`
`conductivity`	`selection`, `charges`, `temperature`	`out`, `device`
`hbond`	`donors`, `acceptors`, `dist_cutoff`	`angle_cutoff`, `out`
`dssp`	—	`out`, `device`
`pca`	`mask`	`n_components`, `out`
`rmsf`	—	`selection`, `out`
`diffusion`	—	`selection`, `out`

See agent_schema.py:78-117 for the complete mapping.

CheckpointConfig

Configuration for progress checkpoints:

class CheckpointConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")
    
    enabled: bool = False
    interval_frames: int = Field(default=1000, ge=1)

Response Schema

RunEnvelope (Union Type)

All responses are one of:

RunEnvelope = Union[RunSuccessEnvelope, RunErrorEnvelope]

Use the status field to discriminate:

"ok" or "dry_run" → RunSuccessEnvelope
"error" → RunErrorEnvelope

RunSuccessEnvelope

class RunSuccessEnvelope(BaseModel):
    schema_version: str = AGENT_RESULT_SCHEMA_VERSION
    status: Literal["ok", "dry_run"]
    exit_code: Literal[0]
    run_id: Optional[str] = None
    output_dir: Optional[str] = None
    system: Optional[Dict[str, Any]] = None
    trajectory: Optional[Dict[str, Any]] = None
    analysis_count: int = Field(ge=0)
    started_at: str
    finished_at: str
    elapsed_ms: int = Field(ge=0)
    warnings: list[str] = Field(default_factory=list)
    results: list[RunResultEntry] = Field(default_factory=list)

Example

{
  "schema_version": "warp-md.agent.v1",
  "status": "ok",
  "exit_code": 0,
  "run_id": "run_20240315_143022",
  "output_dir": "./results",
  "system": {"path": "protein.pdb"},
  "trajectory": {"path": "traj.xtc"},
  "analysis_count": 2,
  "started_at": "2024-03-15T14:30:22Z",
  "finished_at": "2024-03-15T14:30:45Z",
  "elapsed_ms": 23000,
  "warnings": [],
  "results": [
    {
      "analysis": "rg",
      "out": "rg.npz",
      "status": "ok",
      "artifact": {
        "path": "rg.npz",
        "format": "npz",
        "bytes": 4096,
        "sha256": "a1b2c3d4e5f6..."
      }
    }
  ]
}

RunErrorEnvelope

class RunErrorEnvelope(BaseModel):
    schema_version: str = AGENT_RESULT_SCHEMA_VERSION
    status: Literal["error"]
    exit_code: int = Field(ge=1)
    run_id: Optional[str] = None
    output_dir: Optional[str] = None
    system: Optional[Dict[str, Any]] = None
    trajectory: Optional[Dict[str, Any]] = None
    analysis_count: int = Field(ge=0)
    started_at: str
    finished_at: str
    elapsed_ms: int = Field(ge=0)
    warnings: list[str] = Field(default_factory=list)
    results: list[RunResultEntry] = Field(default_factory=list)
    error: RunErrorPayload

Example

{
  "schema_version": "warp-md.agent.v1",
  "status": "error",
  "exit_code": 3,
  "run_id": "run_20240315_143022",
  "analysis_count": 1,
  "started_at": "2024-03-15T14:30:22Z",
  "finished_at": "2024-03-15T14:30:23Z",
  "elapsed_ms": 1000,
  "warnings": [],
  "results": [],
  "error": {
    "code": "E_SELECTION_EMPTY",
    "message": "Selection 'resname XYZ' matched 0 atoms",
    "context": {
      "analysis_index": 0,
      "analysis_name": "rg",
      "selection": "resname XYZ"
    },
    "details": null,
    "traceback": null
  }
}

RunResultEntry

Single analysis result:

class RunResultEntry(BaseModel):
    model_config = ConfigDict(extra="allow")

    analysis: str
    out: str
    status: Literal["ok", "dry_run"]
    artifact: Optional[ArtifactMetadata] = None

ArtifactMetadata

File metadata for result artifacts:

class ArtifactMetadata(BaseModel):
    path: str
    format: str
    bytes: int = Field(ge=0)
    sha256: str = Field(min_length=64, max_length=64)

RunErrorPayload

Error details:

class RunErrorPayload(BaseModel):
    code: str  # One of ErrorCode literals
    message: str
    context: Dict[str, Any]
    details: Optional[Any] = None
    traceback: Optional[str] = None

Error Codes

All error codes are defined as Literal types:

ErrorCode = Literal[
    # Validation errors (exit code 2)
    "E_CONFIG_VALIDATION",
    "E_CONFIG_VERSION",
    "E_CONFIG_MISSING_FIELD",
    
    # Analysis specification errors (exit code 3)
    "E_ANALYSIS_UNKNOWN",
    "E_ANALYSIS_SPEC",
    "E_SELECTION_EMPTY",
    "E_SELECTION_INVALID",
    
    # Runtime errors (exit code 4)
    "E_SYSTEM_LOAD",
    "E_TRAJECTORY_LOAD",
    "E_TRAJECTORY_EOF",
    "E_RUNTIME_EXEC",
    "E_OUTPUT_WRITE",
    "E_DEVICE_UNAVAILABLE",
    
    # Internal errors (exit code 5)
    "E_INTERNAL",
]

Error codes map to exit codes: 2=validation, 3=specification, 4=runtime, 5=internal. This enables agents to handle errors programmatically.

Event Streaming

When stream="ndjson", warp-md emits NDJSON events to stderr:

Event Types

RunEvent = Union[
    RunStartedEvent,
    AnalysisStartedEvent,
    AnalysisCompletedEvent,
    AnalysisFailedEvent,
    RunCompletedEvent,
    RunFailedEvent,
]

RunStartedEvent

class RunStartedEvent(BaseModel):
    event: Literal["run_started"]
    run_id: Optional[str] = None
    config_path: str
    dry_run: bool
    analysis_count: int = Field(ge=0)
    completed: int = Field(ge=0)
    total: int = Field(ge=0)
    progress_pct: float = Field(ge=0.0, le=100.0)
    eta_ms: Optional[int] = Field(default=None, ge=0)

AnalysisStartedEvent

class AnalysisStartedEvent(BaseModel):
    event: Literal["analysis_started"]
    index: int = Field(ge=0)
    analysis: str
    out: str
    completed: int = Field(ge=0)
    total: int = Field(ge=0)
    progress_pct: float = Field(ge=0.0, le=100.0)
    eta_ms: Optional[int] = Field(default=None, ge=0)

AnalysisCompletedEvent

class AnalysisCompletedEvent(BaseModel):
    event: Literal["analysis_completed"]
    index: int = Field(ge=0)
    analysis: str
    status: Literal["ok", "dry_run"]
    out: str
    timing_ms: Optional[int] = Field(default=None, ge=0)
    completed: int = Field(ge=0)
    total: int = Field(ge=0)
    progress_pct: float = Field(ge=0.0, le=100.0)
    eta_ms: Optional[int] = Field(default=None, ge=0)

CheckpointEvent

Emitted periodically during long analyses:

class CheckpointEvent(BaseModel):
    event: Literal["checkpoint"]
    analysis_index: int
    analysis_name: str
    frames_processed: int
    frames_total: Optional[int] = None
    progress_pct: Optional[float] = None
    eta_ms: Optional[int] = None

Enable checkpoints in the request:

{
  "checkpoint": {
    "enabled": true,
    "interval_frames": 1000
  }
}

Analysis Names

All supported analyses are defined as Literal types:

AnalysisName = Literal[
    "rg", "rmsd", "msd", "rotacf",
    "conductivity", "dielectric", "dipole_alignment",
    "ion_pair_correlation", "structure_factor",
    "water_count", "equipartition", "hbond", "rdf",
    "end_to_end", "contour_length", "chain_rg",
    "bond_length_distribution", "bond_angle_distribution",
    "persistence_length", "docking",
    "dssp", "diffusion", "pca", "rmsf", "density",
    "native_contacts", "volmap", "surf", "molsurf",
    "watershell", "tordiff", "projection",
    "gist", "nmr", "jcoupling",
]

Programmatic Validation

Validate Request

from warp_md.agent_schema import validate_run_request
from pydantic import ValidationError

try:
    cfg = validate_run_request({
        "version": "warp-md.agent.v1",
        "system": "protein.pdb",
        "trajectory": "traj.xtc",
        "analyses": [
            {"name": "rg", "selection": "protein"}
        ]
    })
    print(f"Valid! Running {len(cfg.analyses)} analyses")
except ValidationError as exc:
    print(f"Invalid: {exc}")

Generate JSON Schema

from warp_md.agent_schema import (
    run_request_json_schema,
    run_result_json_schema,
    run_event_json_schema,
)
import json

# Get JSON Schema for OpenAPI, LangChain, etc.
request_schema = run_request_json_schema()
print(json.dumps(request_schema, indent=2))

result_schema = run_result_json_schema()
event_schema = run_event_json_schema()

Render Schema (CLI)

# JSON Schema output
warp-md schema request --format json
warp-md schema result --format json
warp-md schema event --format json

# YAML Schema output
warp-md schema request --format yaml

Usage Examples

Python API

from warp_md.runner import run_analyses

result = run_analyses({
    "version": "warp-md.agent.v1",
    "system": "protein.pdb",
    "trajectory": "traj.xtc",
    "device": "cuda",
    "output_dir": "./results",
    "analyses": [
        {"name": "rg", "selection": "protein"},
        {"name": "rmsd", "selection": "backbone"},
    ]
})

if result.status == "ok":
    for r in result.results:
        print(f"✓ {r.analysis}: {r.artifact.path} ({r.artifact.bytes} bytes)")
else:
    print(f"✗ {result.error.code}: {result.error.message}")

CLI with JSON Config

# Create config.json
cat > config.json <<EOF
{
  "version": "warp-md.agent.v1",
  "system": "protein.pdb",
  "trajectory": "traj.xtc",
  "analyses": [
    {"name": "rg", "selection": "protein"},
    {"name": "dssp"}
  ]
}
EOF

# Run with streaming
warp-md run config.json --stream ndjson

MCP Server Integration

The MCP server uses these schemas internally:

# python/warp_md/mcp_server.py
@server.tool()
def run_analysis(
    system_path: str,
    trajectory_path: str,
    analyses: List[Dict[str, Any]],
    output_dir: Optional[str] = None,
    device: str = "auto",
    fail_fast: bool = True,
) -> Dict[str, Any]:
    from .runner import run_analyses
    
    result = run_analyses({
        "version": "warp-md.agent.v1",
        "system": {"path": system_path},
        "trajectory": {"path": trajectory_path},
        "analyses": analyses,
        "fail_fast": fail_fast,
    }, output_dir=output_dir, device=device)
    
    return result.model_dump(mode="json")

Overview

Why warp-md is agent-friendly

Framework Integrations

LangChain, CrewAI, AutoGen examples

Get Started

Core Concepts

Guides

AI Agents

​Schema Version

​Request Schema

​RunRequest

​Fields

​Validation Rules

​AnalysisRequest

​Common Analysis Parameters

​CheckpointConfig

​Response Schema

​RunEnvelope (Union Type)

​RunSuccessEnvelope

​Example

​RunErrorEnvelope

​Example

​RunResultEntry

​ArtifactMetadata

​RunErrorPayload

​Error Codes

​Event Streaming

​Event Types

​RunStartedEvent

​AnalysisStartedEvent

​AnalysisCompletedEvent

​CheckpointEvent

​Analysis Names

​Programmatic Validation

​Validate Request

​Generate JSON Schema

​Render Schema (CLI)

​Usage Examples

​Python API

​CLI with JSON Config

​MCP Server Integration

​See Also

Overview

Framework Integrations

Build docs developers (and LLMs) love

Schema Version

Request Schema

RunRequest

Fields

Validation Rules

AnalysisRequest

Common Analysis Parameters

CheckpointConfig

Response Schema

RunEnvelope (Union Type)

RunSuccessEnvelope

Example

RunErrorEnvelope

Example

RunResultEntry

ArtifactMetadata

RunErrorPayload

Error Codes

Event Streaming

Event Types

RunStartedEvent

AnalysisStartedEvent

AnalysisCompletedEvent

CheckpointEvent

Analysis Names

Programmatic Validation

Validate Request

Generate JSON Schema

Render Schema (CLI)

Usage Examples

Python API

CLI with JSON Config

MCP Server Integration

See Also