Ctx Fenic¶

Ctx Fenic — mvp.ctx_fenic

Cluster: Context & Retrieval | Type: component | MCP Tools: 26

Overview¶

Fenic-style declarative context engineering component for AI agents. Implements relational operators (filter, map_field, join, group_by, aggregate, select, sort) over structured context data, treating context as typed tables you query precisely and combining deterministic transforms with semantic operations in composable pipelines. The core relational operations are suitable for pilot workflows; optional semantic and text operations depend on extra runtime libraries and model/API configuration. Input is validated with a 100k row limit and 1000-char expression limit.

Fenic's core philosophy is offloading context construction (extraction, chunking, retrieval, summarization) out of the agent's prompt window, reducing token consumption in reasoning loops so agents stay focused on decision-making rather than context management.

Pilot-use caveat

For launch and first-user demos, prefer the deterministic core operations: filter, map_field, join, group_by, aggregate, select, sort, pipeline, and dataset/session persistence. Semantic/text operations such as semantic_extract, semantic_classify, embed, cluster, parse_transcript, and chunk_markdown may require optional packages, embeddings support, and model/API credentials. When those dependencies are unavailable, the MCP tools are expected to return clear degraded/unavailable errors rather than silently succeeding.

When to use:

Managing structured context for agent reasoning - memory, retrieval, knowledge bases
Declarative context pipelines: filter relevant facts, join entity data, aggregate summaries
Keeping agent prompt budgets lean by pre-processing context outside the reasoning loop
Framework-agnostic context operations that compose with any agent framework

Example:

from mvp.ctx_fenic import CtxFenicBlock, FenicInput

block = CtxFenicBlock(name="fenic")
result = block.infer(FenicInput(operation="filter", data=[{"x": 1}, {"x": -2}], expression="x > 0"))
# result.unwrap() -> FenicOutput(data=[{"x": 1}], n_rows=1, operation="filter")

Works well with: adapt_pandas, database, ctx_langextract

Public API¶

`FenicRerankDecision`¶

Validated advisory rerank verdict over a returned search-record set.

Field	Type	Default
`ordered_indices`	`tuple[int, ...]`	required
`dropped_indices`	`tuple[int, ...]`	()
`rationale`	`str`	''
`eligible_fingerprint`	`str`	''
`confidence`	`float`	0.0
`degraded`	`bool`	False
`raw_response`	`str`	''

`LLMFenicRerankRuntime`¶

Provider-neutral search-result-rerank runtime backed by G6's LLM caller.

Constructor:

Parameter	Type	Default
`llm`	`LLMCaller \\| None`	None

Methods:

`rerank(query: str, records: list[Any]) -> FenicRerankDecision`¶

`CtxFenicRerankPatternRuntime`¶

Stateless, load-bearing returned-set-ceiling enforcement.

Methods:

`enforce_eligibility(decision: FenicRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxFenicPlanner`¶

Runtime-first advisory result-rerank facade with returned-order fallback.

Constructor:

Parameter	Type	Default
`runtime`	`FenicRerankRuntime \\| None`	None
`pattern_runtime`	`CtxFenicRerankPatternRuntime \\| None`	None

Methods:

`rerank(query: str, records: list[Any]) -> list[Any]`¶

`CtxFenicBlock(AIBlock[FenicInput, FenicOutput, None])`¶

Fenic-style data processing: filter, map_field, join, group_by, aggregate, select, sort.

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

Methods:

`infer(data: FenicInput) -> Result[FenicOutput]`¶

`FenicInput(BaseModel)`¶

Field	Type	Default
`data`	`list[dict[str, Any]]`	required
`operation`	`FenicOp`	required
`field`	`str`	''
`expression`	`str`	''
`output_field`	`str`	''
`join_data`	`list[dict[str, Any]]`	Field(default_factory=list)
`join_key`	`str`	''
`group_by_field`	`str`	''
`agg_function`	`AggFunction`	'count'
`fields`	`list[str]`	Field(default_factory=list)
`ascending`	`bool`	True
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''

Methods:

`data_not_empty(v: list[dict[str, Any]]) -> list[dict[str, Any]]`¶

`FenicOutput(BaseModel)`¶

Field	Type	Default
`data`	`list[dict[str, Any]]`	required
`n_rows`	`int`	required
`operation`	`str`	required
`success`	`bool`	True
`message`	`str`	''
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'completed'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`CtxFenicMCPBlock(AIBlock[MCPFenicInput, MCPFenicOutput, dict])`¶

Full-featured fenic data processing block with SQLite persistence.

Field	Type	Default
`name`	`str`	'ctx_fenic_mcp'
`state`	`dict \\| None`	None
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds \\| None`	None
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)
`agentic_planner`	`CtxFenicPlanner \\| None`	None

Methods:

`infer(data: MCPFenicInput) -> Result[MCPFenicOutput]`¶

`MCPFenicRecord(BaseModel)`¶

Field	Type	Default
`id`	`str`	required
`record_type`	`str`	required
`key`	`str`	required
`value`	`str`	required
`tags`	`list[str]`	Field(default_factory=list)
`timestamp`	`str`	required
`metadata`	`dict[str, Any]`	Field(default_factory=dict)

`MCPFenicInput(BaseModel)`¶

Field	Type	Default
`op`	`MCPFenicOp`	required
`data_json`	`str`	''
`field`	`str`	''
`expression`	`str`	''
`output_field`	`str`	''
`join_data_json`	`str`	''
`join_key`	`str`	''
`group_by_field`	`str`	''
`agg_function`	`str`	'count'
`fields_json`	`str`	''
`ascending`	`bool`	True
`left_data_json`	`str`	''
`right_data_json`	`str`	''
`left_on`	`str`	''
`right_on`	`str`	''
`text_column`	`str`	''
`labels_json`	`str`	''
`n_clusters`	`int`	3
`model`	`str`	''
`transcript`	`str`	''
`markdown`	`str`	''
`chunk_size`	`int`	500
`json_data`	`str`	''
`jq_expression`	`str`	''
`operations_json`	`str`	''
`max_execution_seconds`	`int`	30
`max_tokens_per_minute`	`int`	0
`name`	`str`	''
`description`	`str`	''
`tags`	`list[str]`	Field(default_factory=list)
`session_data_json`	`str`	''
`operations_log_json`	`str`	''
`query`	`str`	''
`top_k`	`int`	5
`limit`	`int`	50
`agentic_rerank`	`bool \\| None`	None
`run_mode`	`Literal['beta', 'production']`	'beta'
`reviewer_signature`	`str`	''
`key`	`str`	''
`value`	`str`	''

`MCPFenicOutput(BaseModel)`¶

Field	Type	Default
`op`	`str`	required
`key`	`str`	''
`value`	`str`	''
`found`	`bool`	False
`count`	`int`	0
`records`	`list[MCPFenicRecord]`	Field(default_factory=list)
`data`	`list[dict[str, Any]]`	Field(default_factory=list)
`n_rows`	`int`	0
`summary`	`str`	''
`message`	`str`	''
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str`	''
`completion_state`	`str`	'completed'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`agentic_evidence`	`dict[str, Any]`	Field(default_factory=dict)

`FenicStore`¶

Sync SQLite fenic store with 2 tables.

Constructor:

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

Methods:

`save_dataset(name: str, description: str, data_json: str, n_rows: int, tags: list[str]) -> str`¶

`load_dataset(name: str) -> dict[str, Any] | None`¶

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

`delete_dataset(name: str) -> bool`¶

`save_session(name: str, description: str, session_data_json: str, operations_log: str, tags: list[str]) -> str`¶

`load_session(name: str) -> dict[str, Any] | None`¶

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

TF-IDF search across datasets and sessions.

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

Functions¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

Decide whether the agentic search-result-rerank planner should be used.

`eligible_fingerprint(records: list[Any]) -> str`¶

sha256 over the returned record texts (order-sensitive).

`validate_fenic_rerank_decision(decision: FenicRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

Returned-set / anti-injection guard for a search-result-rerank decision.

`planner_is_llm_trusted(planner: Any) -> bool`¶

Whether the BLOCK may report llm_used=True for planner.

MCP Tools¶

Operation	Source
`filter`	fenic_mcp
`map_field`	fenic_mcp
`join`	fenic_mcp
`group_by`	fenic_mcp
`aggregate`	fenic_mcp
`select`	fenic_mcp
`sort`	fenic_mcp
`semantic_join`	fenic_mcp
`semantic_extract`	fenic_mcp
`semantic_classify`	fenic_mcp
`embed`	fenic_mcp
`cluster`	fenic_mcp
`parse_transcript`	fenic_mcp
`chunk_markdown`	fenic_mcp
`jq_query`	fenic_mcp
`pipeline`	fenic_mcp
`resource_check`	fenic_mcp
`save_dataset`	fenic_mcp
`load_dataset`	fenic_mcp
`list_datasets`	fenic_mcp
`delete_dataset`	fenic_mcp
`save_session`	fenic_mcp
`load_session`	fenic_mcp
`search`	fenic_mcp
`info`	fenic_mcp
`list_patterns`	fenic_mcp

Ctx Fenic¶

Overview¶

Public API¶

FenicRerankDecision¶

LLMFenicRerankRuntime¶

rerank(query: str, records: list[Any]) -> FenicRerankDecision¶

CtxFenicRerankPatternRuntime¶

enforce_eligibility(decision: FenicRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]¶

CtxFenicPlanner¶

rerank(query: str, records: list[Any]) -> list[Any]¶

CtxFenicBlock(AIBlock[FenicInput, FenicOutput, None])¶

infer(data: FenicInput) -> Result[FenicOutput]¶

FenicInput(BaseModel)¶

data_not_empty(v: list[dict[str, Any]]) -> list[dict[str, Any]]¶

FenicOutput(BaseModel)¶

CtxFenicMCPBlock(AIBlock[MCPFenicInput, MCPFenicOutput, dict])¶

infer(data: MCPFenicInput) -> Result[MCPFenicOutput]¶

MCPFenicRecord(BaseModel)¶

MCPFenicInput(BaseModel)¶

MCPFenicOutput(BaseModel)¶

FenicStore¶

save_dataset(name: str, description: str, data_json: str, n_rows: int, tags: list[str]) -> str¶

load_dataset(name: str) -> dict[str, Any] | None¶

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

delete_dataset(name: str) -> bool¶

save_session(name: str, description: str, session_data_json: str, operations_log: str, tags: list[str]) -> str¶

load_session(name: str) -> dict[str, Any] | None¶

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

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

Functions¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

eligible_fingerprint(records: list[Any]) -> str¶

validate_fenic_rerank_decision(decision: FenicRerankDecision, n_eligible: int, expected_fingerprint: str) -> None¶

planner_is_llm_trusted(planner: Any) -> bool¶

MCP Tools¶

`FenicRerankDecision`¶

`LLMFenicRerankRuntime`¶

`rerank(query: str, records: list[Any]) -> FenicRerankDecision`¶

`CtxFenicRerankPatternRuntime`¶

`enforce_eligibility(decision: FenicRerankDecision, records: list[Any]) -> tuple[list[Any], bool, bool]`¶

`CtxFenicPlanner`¶

`rerank(query: str, records: list[Any]) -> list[Any]`¶

`CtxFenicBlock(AIBlock[FenicInput, FenicOutput, None])`¶

`infer(data: FenicInput) -> Result[FenicOutput]`¶

`FenicInput(BaseModel)`¶

`data_not_empty(v: list[dict[str, Any]]) -> list[dict[str, Any]]`¶

`FenicOutput(BaseModel)`¶

`CtxFenicMCPBlock(AIBlock[MCPFenicInput, MCPFenicOutput, dict])`¶

`infer(data: MCPFenicInput) -> Result[MCPFenicOutput]`¶

`MCPFenicRecord(BaseModel)`¶

`MCPFenicInput(BaseModel)`¶

`MCPFenicOutput(BaseModel)`¶

`FenicStore`¶

`save_dataset(name: str, description: str, data_json: str, n_rows: int, tags: list[str]) -> str`¶

`load_dataset(name: str) -> dict[str, Any] | None`¶

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

`delete_dataset(name: str) -> bool`¶

`save_session(name: str, description: str, session_data_json: str, operations_log: str, tags: list[str]) -> str`¶

`load_session(name: str) -> dict[str, Any] | None`¶

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

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

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

`eligible_fingerprint(records: list[Any]) -> str`¶

`validate_fenic_rerank_decision(decision: FenicRerankDecision, n_eligible: int, expected_fingerprint: str) -> None`¶

`planner_is_llm_trusted(planner: Any) -> bool`¶