Adapt Pandas¶

Adapt Pandas — mvp.adapt_pandas

Cluster: Data Processing | Type: component | MCP Tools: 36

Overview¶

adapt_pandas wraps pandas DataFrame operations (load, profile, transform, filter, aggregate) as AIBlocks, plus a 34-operation MCP sub-package with SQLite persistence for datasets, reusable pipelines, capability discovery, and constrained native adapter operations.

Launch readiness caveat

The component-level review and tests support adapt_pandas as useful for first-user workflows: CSV/JSON/Parquet loading, profiling, filtering, schema checks, pipeline replay, and SQLite-backed dataset reuse all work through the local block and MCP paths. This does not by itself make the full product launch-ready. The launch plan still depends on ecosystem checks outside this component, including clean-machine MCP install validation, dependency pinning, Docker/runtime packaging, staging smoke tests, and non-developer onboarding docs.

When to use:

Data ingestion, profiling, and exploratory analysis
Feature engineering with filter, sort, groupby, merge, and pivot operations
Persisting and replaying named data pipelines via the MCP interface

Example:

from mvp.adapt_pandas import AdaptPandasBlock, DataInput

block = AdaptPandasBlock(name="pd")
result = block.infer(DataInput(records=[{"x": 1}, {"x": 2}]))
# result.ok → True; result.value → DataOutput with shape=(2, 1)

Works well with: adapt_sklearn, adapt_pygad, align_evals, database

Public API¶

`AdaptPandasBlock(AIBlock[DataInput, DataOutput, None])`¶

Applies an optional DataFrame operation to input records and returns results,

Field	Type	Default
`name`	`str`	'adapt_pandas'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`PandasCleaningPlanner \\| None`	None

Methods:

`infer(data: DataInput) -> Result[DataOutput]`¶

`DataOperation(BaseModel)`¶

A single transformation to apply to a DataFrame.

Field	Type	Default
`op`	`Literal['head', 'tail', 'describe', 'dropna', 'reset_index', 'select', 'filter_rows', 'identity', 'read_csv', 'read_parquet', 'read_json', 'concat', 'to_datetime', 'groupby_agg', 'merge', 'join', 'pivot', 'melt', 'recommend']`	required
`params`	`dict[str, Any]`	Field(default_factory=dict)

`DataInput(BaseModel)`¶

Input records plus optional column selection and operation.

Field	Type	Default
`records`	`list[dict[str, Any]]`	required
`columns`	`list[str]`	Field(default_factory=list)
`operation`	`DataOperation \\| None`	None
`data_schema`	`dict[str, str] \\| None`	Field(default=None, alias='schema', serialization_alias='schema')
`run_mode`	`str`	'beta'
`reviewer_signature`	`str`	''

Methods:

`schema() -> dict[str, str] | None`¶

`DataOutput(BaseModel)`¶

Output records plus shape metadata.

Field	Type	Default
`records`	`list[dict[str, Any]]`	required
`shape`	`tuple[int, int]`	required
`columns`	`list[str]`	required
`summary`	`dict[str, Any]`	Field(default_factory=dict)
`warnings`	`list[str]`	Field(default_factory=list)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`ranked_ops`	`list[str]`	Field(default_factory=list)
`recommendation_rationale`	`str`	''
`eligible_ops`	`list[str]`	Field(default_factory=list)
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)
`requires_fallback`	`bool`	False
`diverged_from_floor`	`bool`	False
`deterministic_floor_rank`	`list[str]`	Field(default_factory=list)
`completion_state`	`str`	''

Methods:

`to_input(operation: DataOperation | None = None, schema: dict[str, str] | None = None) -> DataInput`¶

Wrap this output's records back into a DataInput for chaining.

`AdaptPandasMCPBlock(AIBlock[MCPPandasInput, MCPPandasOutput, dict])`¶

28-op pandas MCP block with SQLite persistence.

Field	Type	Default
`name`	`str`	'adapt_pandas_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	field(default_factory=lambda: os.environ.get('PANDAS_DB_PATH', _DEFAULT_DB))
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPPandasInput) -> Result[MCPPandasOutput]`¶

`MCPPandasRecord(BaseModel)`¶

A single metadata record returned from PandasStore queries.

Field	Type	Default
`id`	`str`	required
`record_type`	`str`	required
`name`	`str`	required
`content`	`str`	required
`tags`	`str`	''
`timestamp`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)

`MCPPandasInput(BaseModel)`¶

Input to AdaptPandasMCPBlock — 30 schema ops (28 exposed as MCP tools).

Field	Type	Default
`op`	Literal['ops', 'help', 'head', 'tail', 'describe', 'dropna', 'reset_index', 'select', 'filter_rows', 'sort', 'fillna', 'rename', 'astype', 'groupby_agg', 'merge', 'pivot', 'melt', 'profile', 'value_counts', 'corr', 'dataset_store', 'dataset_retrieve', 'dataset_list', 'dataset_load', 'pipeline_store', 'pipeline_run', 'search', 'info', 'recommend', 'list_patterns', 'native_capability_info', 'native_dataframe_schema', 'native_dataframe_describe', 'native_groupby', 'native_value_counts', 'native_convert_dtypes']	required
`records`	`list[dict]`	Field(default_factory=list)
`columns`	`list[str]`	Field(default_factory=list)
`n`	`int`	5
`by`	`str`	''
`ascending`	`bool`	True
`column`	`str`	''
`value`	`Any`	None
`rename_map`	`dict`	Field(default_factory=dict)
`dtype`	`str`	''
`agg_map`	`dict`	Field(default_factory=dict)
`right_records`	`list[dict]`	Field(default_factory=list)
`on`	`str`	''
`how`	`str`	'inner'
`index`	`str`	''
`values`	`str`	''
`aggfunc`	`str`	'mean'
`id_vars`	`list[str]`	Field(default_factory=list)
`value_vars`	`list[str]`	Field(default_factory=list)
`var_name`	`str`	'variable'
`value_name`	`str`	'value'
`normalize`	`bool`	False
`top_k`	`int`	20
`method`	`str`	'pearson'
`name`	`str`	''
`tags_csv`	`str`	''
`notes`	`str`	''
`limit`	`int`	50
`steps_json`	`str`	''
`description`	`str`	''
`query`	`str`	''
`top_k_results`	`int`	10
`path`	`str`	''
`chunk_size`	`int \\| None`	None
`run_mode`	`str`	'beta'
`reviewer_signature`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPPandasOutput(BaseModel)`¶

Output from AdaptPandasMCPBlock.

Field	Type	Default
`op`	`str`	required
`records`	`list[dict]`	Field(default_factory=list)
`shape`	`list[int]`	Field(default_factory=list)
`columns`	`list[str]`	Field(default_factory=list)
`summary`	`dict`	Field(default_factory=dict)
`name`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`retrieved`	`list[dict]`	Field(default_factory=list)
`scores`	`list[float]`	Field(default_factory=list)
`message`	`str`	''
`metadata`	`dict`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict`	Field(default_factory=dict)
`evidence`	`list[dict]`	Field(default_factory=list)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`PandasStore`¶

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

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`store_dataset(name: str, records: list[dict], tags: str = '', notes: str = '') -> str`¶

Persist a named dataset. Returns the assigned id.

`retrieve_dataset(name: str) -> dict | None`¶

Retrieve a stored dataset by name. Returns None if not found.

`list_datasets(tags_csv: str = '', limit: int = 50) -> list[dict]`¶

List stored datasets, optionally filtered by tags.

`store_pipeline(name: str, steps: list[dict], description: str = '', tags: str = '') -> str`¶

Persist a named pipeline. Returns the assigned id.

`retrieve_pipeline(name: str) -> dict | None`¶

Retrieve a stored pipeline by name. Returns None if not found.

`increment_pipeline_use(name: str) -> None`¶

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

`log_transform(op: str, params: dict, input_shape: str, output_shape: str, success: bool, error_message: str = '', dataset_name: str = '') -> None`¶

`store_profile(dataset_name: str, profile: dict) -> None`¶

`get_profile(dataset_name: str) -> dict | None`¶

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

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

`text_search(query: str, top_k: int = 10) -> list[dict]`¶

Search datasets + pipelines by TF-IDF (falls back to substring).

MCP Tools¶

Operation	Source
`ops`	pandas_mcp
`help`	pandas_mcp
`head`	pandas_mcp
`tail`	pandas_mcp
`describe`	pandas_mcp
`dropna`	pandas_mcp
`reset_index`	pandas_mcp
`select`	pandas_mcp
`filter_rows`	pandas_mcp
`sort`	pandas_mcp
`fillna`	pandas_mcp
`rename`	pandas_mcp
`astype`	pandas_mcp
`groupby_agg`	pandas_mcp
`merge`	pandas_mcp
`pivot`	pandas_mcp
`melt`	pandas_mcp
`profile`	pandas_mcp
`value_counts`	pandas_mcp
`corr`	pandas_mcp
`dataset_store`	pandas_mcp
`dataset_retrieve`	pandas_mcp
`dataset_list`	pandas_mcp
`dataset_load`	pandas_mcp
`pipeline_store`	pandas_mcp
`pipeline_run`	pandas_mcp
`search`	pandas_mcp
`info`	pandas_mcp
`recommend`	pandas_mcp
`list_patterns`	pandas_mcp
`native_capability_info`	pandas_mcp
`native_dataframe_schema`	pandas_mcp
`native_dataframe_describe`	pandas_mcp
`native_groupby`	pandas_mcp
`native_value_counts`	pandas_mcp
`native_convert_dtypes`	pandas_mcp

Adapt Pandas¶

Overview¶

Public API¶

AdaptPandasBlock(AIBlock[DataInput, DataOutput, None])¶

infer(data: DataInput) -> Result[DataOutput]¶

DataOperation(BaseModel)¶

DataInput(BaseModel)¶

schema() -> dict[str, str] | None¶

DataOutput(BaseModel)¶

to_input(operation: DataOperation | None = None, schema: dict[str, str] | None = None) -> DataInput¶

AdaptPandasMCPBlock(AIBlock[MCPPandasInput, MCPPandasOutput, dict])¶

infer(data: MCPPandasInput) -> Result[MCPPandasOutput]¶

MCPPandasRecord(BaseModel)¶

MCPPandasInput(BaseModel)¶

MCPPandasOutput(BaseModel)¶

PandasStore¶

store_dataset(name: str, records: list[dict], tags: str = '', notes: str = '') -> str¶

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

list_datasets(tags_csv: str = '', limit: int = 50) -> list[dict]¶

store_pipeline(name: str, steps: list[dict], description: str = '', tags: str = '') -> str¶

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

increment_pipeline_use(name: str) -> None¶

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

log_transform(op: str, params: dict, input_shape: str, output_shape: str, success: bool, error_message: str = '', dataset_name: str = '') -> None¶

store_profile(dataset_name: str, profile: dict) -> None¶

get_profile(dataset_name: str) -> dict | None¶

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

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

text_search(query: str, top_k: int = 10) -> list[dict]¶

MCP Tools¶

`AdaptPandasBlock(AIBlock[DataInput, DataOutput, None])`¶

`infer(data: DataInput) -> Result[DataOutput]`¶

`DataOperation(BaseModel)`¶

`DataInput(BaseModel)`¶

`schema() -> dict[str, str] | None`¶

`DataOutput(BaseModel)`¶

`to_input(operation: DataOperation | None = None, schema: dict[str, str] | None = None) -> DataInput`¶

`AdaptPandasMCPBlock(AIBlock[MCPPandasInput, MCPPandasOutput, dict])`¶

`infer(data: MCPPandasInput) -> Result[MCPPandasOutput]`¶

`MCPPandasRecord(BaseModel)`¶

`MCPPandasInput(BaseModel)`¶

`MCPPandasOutput(BaseModel)`¶

`PandasStore`¶

`store_dataset(name: str, records: list[dict], tags: str = '', notes: str = '') -> str`¶

`retrieve_dataset(name: str) -> dict | None`¶

`list_datasets(tags_csv: str = '', limit: int = 50) -> list[dict]`¶

`store_pipeline(name: str, steps: list[dict], description: str = '', tags: str = '') -> str`¶

`retrieve_pipeline(name: str) -> dict | None`¶

`increment_pipeline_use(name: str) -> None`¶

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

`log_transform(op: str, params: dict, input_shape: str, output_shape: str, success: bool, error_message: str = '', dataset_name: str = '') -> None`¶

`store_profile(dataset_name: str, profile: dict) -> None`¶

`get_profile(dataset_name: str) -> dict | None`¶

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

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

`text_search(query: str, top_k: int = 10) -> list[dict]`¶