Adapt Physical Ai¶

Adapt Physical AI — mvp.adapt_physical_ai

Cluster: Physical AI | Type: component | MCP Tools: 32

Overview¶

Physics simulation and motion-planning block that models rigid-body dynamics, collision detection, path planning (A* and RRT), forward/inverse kinematics for jointed robots, and simulated device command dispatch. Maintains a stateful physics world across calls — position, velocity, obstacles, gravity — so a trajectory can be built up step by step. Simulation state can be saved to and loaded from named snapshots, including through the MCP block's SQLite-backed store.

The MCP surface also exposes constrained backend-native adapter ops (backend_capabilities, backend_describe_op, and backend_call). These are explicit, allowlisted inspection/call surfaces for the bundled simulation, motion-planning, and in-memory device-driver backends. They are not arbitrary passthroughs and do not enable real hardware control. Missing transports, absent optional backends, simulation-only execution, advisory planning, and blocked hardware actuation are surfaced with completion_state, warning_card, evidence, and correlation IDs when supplied.

When to use:

• Simulating robot or agent trajectories before any separate, reviewed hardware integration step
• Planning collision-free paths through a 2D grid or continuous space using A* or RRT
• Computing forward kinematics for joint-angle sequences or solving inverse kinematics targets
• Prototyping a device-driver interface with the bundled in-memory simulation backend

Example:

from mvp.adapt_physical_ai import AdaptPhysicalAIBlock, PhysicalAIInput

block = AdaptPhysicalAIBlock(name="phys_ai")
result = block.infer(PhysicalAIInput(
    op="plan_path_astar",
    start=[0, 0],
    goal=[9, 9],
    grid_size=[10, 10],
    obstacles=[{"x": 3, "y": 3}, {"x": 3, "y": 4}],
))
# result.value.path → list of [x, y] waypoints from start to goal

Works well with: adapt_synthetic_world, adapt_blender, goal_engine

Public API¶

DeviceDriver(Protocol)¶

Minimal interface a device backend must satisfy.

Methods:

connect(device_id: str) -> dict¶

Open a connection to device_id. Return status dict.

send_command(device_id: str, command: str) -> dict¶

Send command to an already-connected device_id. Return result dict.

get_status(device_id: str) -> dict¶

Return current status dict for device_id.

disconnect(device_id: str) -> dict¶

Close the connection to device_id. Return status dict.

SimDeviceDriver¶

Pure in-memory simulation driver (default).

Methods:

connect(device_id: str) -> dict¶

send_command(device_id: str, command: str) -> dict¶

get_status(device_id: str) -> dict¶

disconnect(device_id: str) -> dict¶

AdaptPhysicalAIBlock(AIBlock[PhysicalAIInput, PhysicalAIOutput, dict])¶

Physics simulation and motion planning block (27 ops).

Methods:

infer(data: PhysicalAIInput) -> Result[PhysicalAIOutput]¶

PhysicalAIInput(BaseModel)¶

Input to AdaptPhysicalAIBlock.

PhysicalAIOutput(BaseModel)¶

Output from AdaptPhysicalAIBlock.

AdaptPhysicalAIMCPBlock(AIBlock[MCPPhysicalAIInput, MCPPhysicalAIOutput, dict])¶

27-op physical AI MCP block with SQLite persistence.

Methods:

infer(data: MCPPhysicalAIInput) -> Result[MCPPhysicalAIOutput]¶

MCPPhysicalAIInput(BaseModel)¶

Input to AdaptPhysicalAIMCPBlock — 27 ops.

MCPPhysicalAIOutput(BaseModel)¶

Output from AdaptPhysicalAIMCPBlock.

PhysicalAIStore¶

SQLite-backed store for the adapt_physical_ai MCP sub-package.

Constructor:

Methods:

save_simulation(name: str, state: dict, tags: str = '') -> str¶

load_simulation(name: str) -> dict | None¶

list_simulations(limit: int = 50) -> list[dict]¶

store_trajectory(sim_name: str, trajectory: list, steps: int, sim_time: float) -> str¶

add_obstacle(position: list, radius: float = 0.5, tags: str = '') -> str¶

list_obstacles(limit: int = 100) -> list[dict]¶

remove_obstacle(obstacle_id: str) -> bool¶

register_device(device_id: str, config: dict | None = None) -> str¶

log_command(device_id: str, command: str, result: str = '') -> None¶

save_state(name: str, state: dict) -> str¶

load_state(name: str) -> dict | None¶

log_error(op: str, error_message: str, params: dict | None = None) -> None¶

count_all() -> dict[str, int]¶

MCP Tools¶