extract_from_column
Extract specific patterns from CSV columns using regex capturing groups. Parse email addresses, product codes, names, dates, and other structured data into separate columns for analysis.
Instructions
Extract patterns from a column using regex with capturing groups.
Returns: ColumnOperationResult with extraction details
Examples: # Extract email parts extract_from_column(ctx, "email", r"(.+)@(.+)")
# Extract code components
extract_from_column(ctx, "product_code", r"([A-Z]{2})-(\d+)")
# Extract and expand into multiple columns
extract_from_column(ctx, "full_name", r"(\w+)\s+(\w+)", expand=True)
# Extract year from date string
extract_from_column(ctx, "date", r"\d{4}")Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| column | Yes | Column name to extract patterns from | |
| pattern | Yes | Regex pattern with capturing groups to extract | |
| expand | Yes | Whether to expand multiple groups into separate columns |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| success | No | Whether operation completed successfully | |
| operation | Yes | Type of operation performed | |
| transform | No | Transform description | |
| part_index | No | Part index for split operations | |
| nulls_filled | No | Number of null values filled | |
| rows_removed | No | Number of rows removed (for remove_duplicates) | |
| rows_affected | Yes | Number of rows affected by operation | |
| values_filled | No | Number of values filled (for fill_missing_values) | |
| updated_sample | No | Sample values after operation | |
| original_sample | No | Sample values before operation | |
| columns_affected | Yes | Names of columns affected |
Implementation Reference
- The main async handler function implementing the 'extract_from_column' tool. It validates the regex pattern, extracts matches from the specified column using pandas str.extract, optionally expands capturing groups into new columns, and returns a ColumnOperationResult.
async def extract_from_column( ctx: Annotated[Context, Field(description="FastMCP context for session access")], column: Annotated[str, Field(description="Column name to extract patterns from")], pattern: Annotated[str, Field(description="Regex pattern with capturing groups to extract")], *, expand: Annotated[ bool, Field(description="Whether to expand multiple groups into separate columns"), ] = False, ) -> ColumnOperationResult: r"""Extract patterns from a column using regex with capturing groups. Returns: ColumnOperationResult with extraction details Examples: # Extract email parts extract_from_column(ctx, "email", r"(.+)@(.+)") # Extract code components extract_from_column(ctx, "product_code", r"([A-Z]{2})-(\d+)") # Extract and expand into multiple columns extract_from_column(ctx, "full_name", r"(\w+)\s+(\w+)", expand=True) # Extract year from date string extract_from_column(ctx, "date", r"\d{4}") """ # Get session_id from FastMCP context session_id = ctx.session_id _session, df = get_session_data(session_id) _validate_column_exists(column, df) # Validate regex pattern try: re.compile(pattern) except re.error as e: msg = "pattern" raise InvalidParameterError(msg, pattern, f"Invalid regex pattern: {e}") from e # Apply extraction # pandas typing limitation: str.extract(expand=bool) overload not properly typed in pandas-stubs # See: https://github.com/pandas-dev/pandas-stubs/issues/43 extracted = df[column].astype(str).str.extract(pattern, expand=expand) # type: ignore[call-overload] if expand and isinstance(extracted, pd.DataFrame): # Multiple capturing groups - create new columns columns_created = [] for i in range(len(extracted.columns)): new_col_name = f"{column}_extracted_{i}" df[new_col_name] = extracted.iloc[:, i] columns_created.append(new_col_name) affected_columns = columns_created operation_desc = f"extract_expand_{len(columns_created)}_groups" else: # Single group or no expand - replace original column if isinstance(extracted, pd.DataFrame): # Multiple groups but not expanding - take first group df[column] = extracted.iloc[:, 0] else: # Single series result df[column] = extracted affected_columns = [column] operation_desc = "extract_pattern" # Count successful extractions (non-null results) if expand and isinstance(extracted, pd.DataFrame): successful_extractions = int((~extracted.isna()).any(axis=1).sum()) else: successful_extractions = ( int(extracted.notna().sum()) if hasattr(extracted, "notna") else len(extracted) ) return ColumnOperationResult( operation=operation_desc, rows_affected=successful_extractions, columns_affected=affected_columns, ) - src/databeak/servers/column_text_server.py:532-532 (registration)Registers the extract_from_column handler as a FastMCP tool named 'extract_from_column' on the column_text_server instance.
column_text_server.tool(name="extract_from_column")(extract_from_column) - Pydantic schema definitions for tool parameters via Annotated types and Field descriptions.
async def extract_from_column( ctx: Annotated[Context, Field(description="FastMCP context for session access")], column: Annotated[str, Field(description="Column name to extract patterns from")], pattern: Annotated[str, Field(description="Regex pattern with capturing groups to extract")], *, expand: Annotated[ bool, Field(description="Whether to expand multiple groups into separate columns"), ] = False, ) -> ColumnOperationResult: