Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions codenib/wiki/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,9 +25,12 @@
search_visual_context,
)
from .media_pipeline import build_multimodal_repository_knowledge
from .media_tools import MULTIMODAL_TOOL_SCHEMAS, MultimodalKnowledgeToolRouter
from .media_vlm import OpenAICompatibleVisualFactExtractor

__all__ = [
"MULTIMODAL_TOOL_SCHEMAS",
"MultimodalKnowledgeToolRouter",
"WikiBuilder",
"OpenAICompatibleVisualFactExtractor",
"build_media_evidence_pack",
Expand Down
138 changes: 138 additions & 0 deletions codenib/wiki/media_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors
#
# SPDX-License-Identifier: Apache-2.0

"""MCP-compatible query surface for multimodal repository knowledge."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Mapping

from .media_knowledge import (
find_visual_code_links,
get_visual_evidence,
search_visual_context,
)

_MAX_QUERY_BYTES = 4096
_MAX_PATH_BYTES = 4096
_MAX_LIMIT = 20

MULTIMODAL_TOOL_SCHEMAS: tuple[dict[str, Any], ...] = (
{
"name": "search_visual_context",
"description": "Search repository visual artifacts, facts, and source bindings.",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": _MAX_LIMIT},
},
"required": ["query"],
"additionalProperties": False,
},
},
{
"name": "get_visual_evidence",
"description": "Return one visual evidence entry by repository-relative artifact path.",
"input_schema": {
"type": "object",
"properties": {"artifact_path": {"type": "string"}},
"required": ["artifact_path"],
"additionalProperties": False,
},
},
{
"name": "find_visual_code_links",
"description": "Find visual artifacts grounded to a source file and optional symbol.",
"input_schema": {
"type": "object",
"properties": {
"source_path": {"type": "string"},
"symbol": {"type": "string"},
},
"required": ["source_path"],
"additionalProperties": False,
},
},
)


@dataclass(frozen=True)
class MultimodalKnowledgeToolRouter:
"""Small tool router that mirrors the future MCP surface."""

view: Mapping[str, Any]

def tool_schemas(self) -> list[dict[str, Any]]:
return [dict(schema) for schema in MULTIMODAL_TOOL_SCHEMAS]

def call_tool(self, name: str, arguments: Mapping[str, Any]) -> dict[str, Any]:
if not isinstance(arguments, Mapping):
raise ValueError("multimodal tool arguments must be an object")
if name == "search_visual_context":
query = _bounded_text(arguments.get("query"), label="query")
limit = _limit(arguments.get("limit", 5))
return {
"results": search_visual_context(self.view, query, limit=limit),
}
if name == "get_visual_evidence":
artifact_path = _bounded_text(
arguments.get("artifact_path"),
label="artifact_path",
max_bytes=_MAX_PATH_BYTES,
)
return {"evidence": get_visual_evidence(self.view, artifact_path)}
if name == "find_visual_code_links":
source_path = _bounded_text(
arguments.get("source_path"),
label="source_path",
max_bytes=_MAX_PATH_BYTES,
)
symbol = _bounded_text(
arguments.get("symbol", ""),
label="symbol",
max_bytes=_MAX_PATH_BYTES,
allow_empty=True,
)
return {
"links": find_visual_code_links(
self.view,
source_path,
symbol=symbol,
)
}
raise ValueError(f"unknown multimodal knowledge tool: {name}")


def _bounded_text(
value: Any,
*,
label: str,
max_bytes: int = _MAX_QUERY_BYTES,
allow_empty: bool = False,
) -> str:
text = str(value or "").strip()
if not text and not allow_empty:
raise ValueError(f"{label} is required")
if len(text.encode("utf-8")) > max_bytes:
raise ValueError(f"{label} exceeds the byte limit")
if any(ord(character) < 0x20 for character in text):
raise ValueError(f"{label} contains control characters")
return text


def _limit(value: Any) -> int:
if isinstance(value, bool):
raise ValueError("limit must be an integer")
try:
limit = int(value)
except (TypeError, ValueError) as exc:
raise ValueError("limit must be an integer") from exc
if not 1 <= limit <= _MAX_LIMIT:
raise ValueError(f"limit must be between 1 and {_MAX_LIMIT}")
return limit


__all__ = ["MULTIMODAL_TOOL_SCHEMAS", "MultimodalKnowledgeToolRouter"]
4 changes: 4 additions & 0 deletions docs/experiments/multimodal_repository_knowledge.md
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,10 @@ a queryable view. It exposes three functions that future MCP tools can wrap:
- `get_visual_evidence`
- `find_visual_code_links`

`codenib.wiki.media_tools.MultimodalKnowledgeToolRouter` exposes the same
surface as an MCP-compatible tool router with stable tool schemas and bounded
input validation. This keeps the query surface testable before wiring it into a
server-specific MCP registration path.
## Why evidence stays server-side

Media generation may use bounded source snippets inside provider prompts. Those
Expand Down
109 changes: 109 additions & 0 deletions test/wiki/test_media_tools.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
# SPDX-FileCopyrightText: 2025-2026 CodeNib Contributors
#
# SPDX-License-Identifier: Apache-2.0

from __future__ import annotations

import pytest

from codenib.wiki.media_tools import (
MULTIMODAL_TOOL_SCHEMAS,
MultimodalKnowledgeToolRouter,
)


def _view():
return {
"entries": [
{
"artifact": {
"path": "docs/architecture.svg",
"caption": "IndexCompiler architecture",
"role_hint": "architecture_diagram",
},
"facts": {
"entities": [{"name": "IndexCompiler", "type": "component"}],
"claims": [{"text": "IndexCompiler writes to VectorStore."}],
},
"bindings": [
{
"artifact_path": "docs/architecture.svg",
"entity_name": "IndexCompiler",
"source_path": "codenib/compiler/index_compiler.py",
"symbol": "IndexCompiler",
"kind": "symbol",
"line": 42,
"score": 1.0,
"evidence": "exact symbol match",
}
],
"search_text": (
"docs/architecture.svg IndexCompiler architecture "
"codenib/compiler/index_compiler.py"
),
}
]
}


def test_multimodal_tool_schemas_are_exposed():
names = {schema["name"] for schema in MULTIMODAL_TOOL_SCHEMAS}

assert names == {
"search_visual_context",
"get_visual_evidence",
"find_visual_code_links",
}


def test_tool_router_searches_visual_context():
router = MultimodalKnowledgeToolRouter(_view())

result = router.call_tool(
"search_visual_context",
{"query": "IndexCompiler", "limit": 1},
)

assert result["results"][0]["artifact_path"] == "docs/architecture.svg"


def test_tool_router_gets_visual_evidence():
router = MultimodalKnowledgeToolRouter(_view())

result = router.call_tool(
"get_visual_evidence",
{"artifact_path": "docs/architecture.svg"},
)

assert result["evidence"]["artifact"]["caption"] == "IndexCompiler architecture"


def test_tool_router_finds_visual_code_links():
router = MultimodalKnowledgeToolRouter(_view())

result = router.call_tool(
"find_visual_code_links",
{
"source_path": "codenib/compiler/index_compiler.py",
"symbol": "IndexCompiler",
},
)

assert result["links"][0]["binding"]["line"] == 42


@pytest.mark.parametrize(
("name", "arguments", "message"),
[
("unknown", {}, "unknown"),
("search_visual_context", {"query": ""}, "query"),
("search_visual_context", {"query": "x", "limit": 100}, "limit"),
("get_visual_evidence", {"artifact_path": "bad\npath"}, "control"),
("find_visual_code_links", {"source_path": ""}, "source_path"),
],
)
def test_tool_router_validates_inputs(name, arguments, message):
router = MultimodalKnowledgeToolRouter(_view())

with pytest.raises(ValueError, match=message):
router.call_tool(name, arguments)
Loading