feat: add incremental OOT block path management and auto-discovery

Add add_block_path() and get_block_paths() MCP tools for incremental
OOT module loading with BlockPathsModel responses. On startup, auto-scan
/usr/local/share and ~/.local/share for OOT blocks so modules like
gr-lora_sdr are available without manual configuration.
This commit is contained in:
Ryan Malloy 2026-01-30 18:02:27 -07:00
parent 6dffd936ae
commit dca4e80857
6 changed files with 111 additions and 2 deletions

23
main.py
View File

@ -1,11 +1,16 @@
from __future__ import annotations
import logging
import os
from fastmcp import FastMCP
from gnuradio_mcp.middlewares.platform import PlatformMiddleware
from gnuradio_mcp.providers.mcp import McpPlatformProvider
from gnuradio_mcp.providers.mcp_runtime import McpRuntimeProvider
logger = logging.getLogger(__name__)
try:
from gnuradio import gr
from gnuradio.grc.core.platform import Platform
@ -21,7 +26,23 @@ platform.build_library()
app: FastMCP = FastMCP("GNU Radio MCP", instructions="Create GNU Radio flowgraphs")
McpPlatformProvider.from_platform_middleware(app, PlatformMiddleware(platform))
pmw = PlatformMiddleware(platform)
# Auto-discover OOT modules from common install locations
oot_candidates = [
"/usr/local/share/gnuradio/grc/blocks",
os.path.expanduser("~/.local/share/gnuradio/grc/blocks"),
]
for path in oot_candidates:
if os.path.isdir(path):
try:
result = pmw.add_block_path(path)
if result.blocks_added > 0:
logger.info(f"OOT: +{result.blocks_added} blocks from {path}")
except Exception:
pass
McpPlatformProvider.from_platform_middleware(app, pmw)
McpRuntimeProvider.create(app)
if __name__ == "__main__":

View File

@ -8,7 +8,7 @@ from gnuradio.grc.core.platform import Platform
from gnuradio_mcp.middlewares.base import ElementMiddleware
from gnuradio_mcp.middlewares.flowgraph import FlowGraphMiddleware
from gnuradio_mcp.models import BlockTypeDetailModel, BlockTypeModel
from gnuradio_mcp.models import BlockPathsModel, BlockTypeDetailModel, BlockTypeModel
class PlatformMiddleware(ElementMiddleware):
@ -88,6 +88,36 @@ class PlatformMiddleware(ElementMiddleware):
"blocks_after": blocks_after,
}
def _rebuild_library(self) -> int:
"""Rebuild block library with default + OOT paths. Returns block count."""
all_paths = self.default_block_paths + self._oot_paths
self._platform.build_library(path=all_paths)
return len(self._platform.blocks)
def add_block_path(self, path: str) -> BlockPathsModel:
"""Add a directory of block YAMLs and rebuild the library."""
path = os.path.expanduser(os.path.abspath(path))
if not os.path.isdir(path):
raise FileNotFoundError(f"Block path not found: {path}")
if path in self._oot_paths:
return self.get_block_paths()
before = len(self._platform.blocks)
self._oot_paths.append(path)
total = self._rebuild_library()
return BlockPathsModel(
paths=self._oot_paths.copy(),
block_count=total,
blocks_added=total - before,
)
def get_block_paths(self) -> BlockPathsModel:
"""Return current OOT paths and block count."""
return BlockPathsModel(
paths=self._oot_paths.copy(),
block_count=len(self._platform.blocks),
)
def make_flowgraph(self, filepath: str = "") -> FlowGraphMiddleware:
return FlowGraphMiddleware.from_file(self, filepath)

View File

@ -355,3 +355,11 @@ class EmbeddedBlockIOModel(BaseModel):
sources: list[tuple[str, str, int]]
doc: str = ""
callbacks: list[str] = []
class BlockPathsModel(BaseModel):
"""Result of block path operations."""
paths: list[str]
block_count: int
blocks_added: int = 0

View File

@ -5,6 +5,7 @@ from gnuradio_mcp.models import (
SINK,
SOURCE,
BlockModel,
BlockPathsModel,
BlockTypeDetailModel,
BlockTypeModel,
ConnectionModel,
@ -129,6 +130,18 @@ class PlatformProvider:
"""
return self._platform_mw.load_oot_paths(paths)
def add_block_path(self, path: str) -> BlockPathsModel:
"""Add a directory containing OOT module block YAML files.
Rebuilds the block library to include blocks from the new path.
Use this to load OOT modules like gr-lora_sdr, gr-osmosdr, etc.
"""
return self._platform_mw.add_block_path(path)
def get_block_paths(self) -> BlockPathsModel:
"""Show current OOT block paths and total block count."""
return self._platform_mw.get_block_paths()
##############################################
# Gap 1: Code Generation
##############################################

View File

@ -33,6 +33,8 @@ class McpPlatformProvider:
# ── OOT Block Loading ──────────────────
t(p.load_oot_blocks)
t(p.add_block_path)
t(p.get_block_paths)
# ── Gap 1: Code Generation ─────────────
t(p.generate_code)

View File

@ -11,6 +11,7 @@ from gnuradio_mcp.middlewares.flowgraph import FlowGraphMiddleware
from gnuradio_mcp.middlewares.platform import PlatformMiddleware
from gnuradio_mcp.models import (
BlockModel,
BlockPathsModel,
BlockTypeDetailModel,
FlowgraphOptionsModel,
GeneratedCodeModel,
@ -263,3 +264,37 @@ def test_generate_code_default_output_persists(
main = next((f for f in result.files if f.is_main), None)
if main:
assert os.path.exists(os.path.join(result.output_dir, main.filename))
# ──────────────────────────────────────────────
# OOT Block Path Management
# ──────────────────────────────────────────────
def test_get_block_paths(platform_middleware: PlatformMiddleware):
result = platform_middleware.get_block_paths()
assert isinstance(result, BlockPathsModel)
assert isinstance(result.paths, list)
assert result.block_count > 0
def test_add_block_path_nonexistent_raises(platform_middleware: PlatformMiddleware):
with pytest.raises(FileNotFoundError):
platform_middleware.add_block_path("/nonexistent/path")
def test_add_block_path_idempotent(platform_middleware: PlatformMiddleware, tmp_path):
result = platform_middleware.add_block_path(str(tmp_path))
assert str(tmp_path) in result.paths
result2 = platform_middleware.add_block_path(str(tmp_path))
assert result2.paths.count(str(tmp_path)) == 1
def test_add_block_path_returns_block_count(
platform_middleware: PlatformMiddleware, tmp_path
):
result = platform_middleware.add_block_path(str(tmp_path))
assert isinstance(result, BlockPathsModel)
assert result.block_count > 0
# Empty dir won't add new blocks, but count stays the same
assert result.blocks_added >= 0