Skip to content

Ctx Vision

ctx_vision -- mvp.ctx_vision

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

Overview

Multimodal vision block that sends images (base64-encoded, data URL, or URL) to a vision-capable LLM via OpenRouter/litellm, with the model defaulting to google/gemini-2.5-flash (overridable via VISION_MODEL env var or VisionInput.model). Five core operations are supported: describe, answer_question, extract_text (OCR), classify, and detect_objects; the MCP package adds 20 FastMCP data-plane tools plus the off-plane list_patterns dispatcher op, with SQLite-backed analysis storage. OpenRouter calls require an API key and explicit cost consent via allow_paid_api=True or ALLOW_PAID_API=1; without that setup, image operations fail with an actionable error. Successful VLM-derived outputs carry the canonical completion_state, warning_card, and evidence envelope as qualified-draft (VLM output is inherently unverified and review-required; it is never auto-promoted to verified); comparison uses comparison_method="description_overlap" with pixel_similarity_validated=False, and object detection uses detection_method="vlm_text_parse". Text-bearing ops (describe, answer_question, extract_text) treat their extracted/described text as untrusted image-origin content: the output carries evidence.content_provenance="untrusted_image_extract", the text is scanned with the shared mvp.csf_cognitive.adversarial.detect_injection detector, and a matched prompt-injection pattern surfaces a CTX_VISION_INJECTION_IN_EXTRACTED_TEXT warning card while the legitimate result is retained at qualified-draft (annotate, never block or blank).

When to use:

  • Extracting structured information from images in a physical AI or document pipeline
  • Running OCR on scanned documents or screenshots without a dedicated OCR service
  • Classifying or labelling images as part of an automated data curation workflow

Example:

from mvp.ctx_vision import CtxVisionBlock, VisionInput

block = CtxVisionBlock(name="vision")
result = block.infer(VisionInput(
    op="describe",
    image_url="https://example.com/img.png",
    allow_paid_api=True,
))
# result.value.description -> LLM-generated description; result.value.backend -> "llm"

Works well with: ctx_markitdown, multimodal, sensory_fusion

Public API

VisionInput(BaseModel)

Input to CtxVisionBlock.

Field Type Default
op str 'describe'
run_mode Literal['beta', 'production'] 'beta'
reviewer_signature str ''
image_base64 str ''
image_url str ''
question str ''
prompt str ''
model str ''
max_tokens int 300
allow_paid_api bool False

Methods:

model_post_init(__context) -> None

Validate that image ops have an image source.

VisionOutput(BaseModel)

Output from CtxVisionBlock.

Field Type Default
op str ''
description str ''
text str ''
labels list[str] Field(default_factory=list)
objects list[dict] Field(default_factory=list)
confidence float 0.0
model_used str ''
tokens_used int 0
backend str ''
degraded bool False
degradation_reason str \| None None
available_with str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
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 ''
confidence_meaning str 'unknown'
review_required bool True
description_overlap_score float 0.0
pixel_similarity_validated bool False

CtxVisionSkill

Field Type Default
name str required
pattern_slug str required
description str required
executable bool required
mechanism str required
capabilities tuple[str, ...] required
triggers tuple[str, ...] required
risk_notes tuple[str, ...] required

Methods:

compact() -> dict[str, Any]

CtxVisionSkillCatalog

Maps each applied pattern slug to a ctx_vision deterministic-safety skill.

Methods:

list_skills() -> list[CtxVisionSkill]

executable_skills() -> list[CtxVisionSkill]

get(slug: str) -> CtxVisionSkill | None

CtxVisionBlock(AIBlock[VisionInput, VisionOutput, None])

Multimodal vision block.

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

Methods:

infer(data: VisionInput) -> Result[VisionOutput]

MCPVisionInput(BaseModel)

Input for any of the 20 vision MCP operations.

Field Type Default
op VisionOp required
image_base64 str ''
image_url str ''
image_base64_b str ''
image_url_b str ''
question str ''
prompt str ''
model str ''
max_tokens int 300
allow_paid_api bool False
images_json str ''
analysis_id str ''
tags str ''
limit int 50
description str ''
text str ''
labels_json str ''
objects_json str ''
confidence float 0.0
model_used str ''
tokens_used int 0
backend str ''
metadata_json str ''
context str ''
run_mode Literal['beta', 'production'] 'beta'
reviewer_signature str ''

MCPVisionOutput(BaseModel)

Output from any of the 20 vision MCP operations.

Field Type Default
op str ''
description str ''
text str ''
labels list[str] Field(default_factory=list)
objects list[dict] Field(default_factory=list)
confidence float 0.0
confidence_meaning str 'unknown'
review_required bool True
model_used str ''
tokens_used int 0
backend str ''
differences list[str] Field(default_factory=list)
similarity float 0.0
description_overlap_score float 0.0
pixel_similarity_validated bool False
batch_results list[dict] Field(default_factory=list)
analysis_id str ''
found bool False
count int 0
records list[dict] Field(default_factory=list)
message str ''
summary str ''
metadata dict[str, Any] Field(default_factory=dict)
degraded bool False
degradation_reason str ''
completion_state Literal['verified', 'qualified-draft', 'blocked-escalated'] 'qualified-draft'
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 ''
patterns list[dict] Field(default_factory=list)

VisionStore

SQLite-backed store for vision analysis results.

Constructor:

Parameter Type Default
db_path str ':memory:'

Methods:

save_analysis(op: str = '', description: str = '', text: str = '', labels: list[str] | None = None, objects: list[dict] | None = None, confidence: float = 0.0, model_used: str = '', tokens_used: int = 0, backend: str = '', tags: list[str] | None = None, metadata: dict[str, Any] | None = None, analysis_id: str = '') -> str

get_analysis(analysis_id: str) -> dict[str, Any] | None

list_analyses(op: str = '', tags: list[str] | None = None, limit: int = 50) -> list[dict[str, Any]]

delete_analysis(analysis_id: str) -> bool

count_all() -> dict[str, int]

stats() -> dict[str, Any]

CtxVisionMCPBlock(AIBlock[MCPVisionInput, MCPVisionOutput, dict])

20-op vision MCP block with SQLite persistence (+ list_patterns introspection).

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

Methods:

store() -> VisionStore

core() -> CtxVisionBlock

infer(data: MCPVisionInput) -> Result[MCPVisionOutput]

Functions

applied_agentic_patterns() -> list[dict[str, Any]]

Return compact metadata for ctx_vision-applied vendored patterns.

get_skill_catalog() -> CtxVisionSkillCatalog

MCP Tools

Operation Source
describe vision_mcp
answer_question vision_mcp
extract_text vision_mcp
classify vision_mcp
detect_objects vision_mcp
analyze_screenshot vision_mcp
read_diagram vision_mcp
parse_table vision_mcp
summarize_document vision_mcp
compare_images vision_mcp
find_differences vision_mcp
similarity_score vision_mcp
batch_describe vision_mcp
batch_classify vision_mcp
batch_extract vision_mcp
save_analysis vision_mcp
list_analyses vision_mcp
get_analysis vision_mcp
list_models vision_mcp
get_info vision_mcp
list_patterns vision_mcp