Add knowledge-mapper run CLI subcommand (#10)

CONTEXT.md

@@ -11,7 +11,7 @@
 1. **A Python SDK** — write Python code, use decorators, implement your own handlers.
 2. **A settings/config-driven approach** (work in progress) — for simple cases where no custom handler logic is needed, a config file (YAML/JSON) plus a CLI should suffice.
 
-Both approaches are planned and partially implemented. The SDK path is fully functional; the config-driven CLI path is tracked under issue #10.
+Both approaches are planned and partially implemented. The SDK path is fully functional, and the `knowledge-mapper run path/to/file.py:kb` CLI starts an SDK-defined KB (issue #10). A future config-driven CLI (loading a `KnowledgeBase` from a YAML/JSON settings file) is tracked separately.
 
 ---
 
@@ -418,7 +418,7 @@ The pre-overhaul, config-file-driven mapper implementation has been removed from
 |---|-------|-------|
 | [#5](https://github.com/TNO/knowledge-mapper/issues/5) | Make `result_pattern` optional for POST interactions | `PostReactInteractionInfo.result_graph_pattern` is currently required; should be optional |
 | [#6](https://github.com/TNO/knowledge-mapper/issues/6) | Allow domain knowledge loading via KE client | Extend `Client`/`ClientProtocol` with methods to load domain knowledge into the SC (supported since KE 1.3.1) |
-| [#10](https://github.com/TNO/knowledge-mapper/issues/10) | Create simple CLI for starting KM | Entry point for the config-driven approach; part of the "both SDK and config-driven" roadmap |
+| [#10](https://github.com/TNO/knowledge-mapper/issues/10) | Create simple CLI for starting KM | ✅ Done — `knowledge-mapper run path/to/file.py:kb` (typer-based, see `src/knowledge_mapper/cli.py`) |
 | [#11](https://github.com/TNO/knowledge-mapper/issues/11) | Add dependency injection system | Allow handlers to declare dependencies (e.g. DB connections) that are injected at call time |
 | [#15](https://github.com/TNO/knowledge-mapper/issues/15) | Create default handlers for POST and ASK interactions | ASK/POST KIs are now auto-registered from settings via `builder.build()` with no handler; outgoing-only KIs need no handler |
 | [#23](https://github.com/TNO/knowledge-mapper/issues/23) | Extract settings-based KI registration out of KnowledgeBase | ✅ Done — moved to `KnowledgeBaseBuilder` in `src/knowledge_base_builder.py` |

README.md

@@ -50,6 +50,16 @@ For local development (from the repository root):
 pip install -e .
 ```
 
+## CLI
+
+Installing the package also provides a `knowledge-mapper` command. Use `run` to start a `KnowledgeBase` defined in a Python file, replacing the `asyncio.run(...)` boilerplate:
+
+```bash
+knowledge-mapper run my_app.py:kb
+```
+
+Where `my_app.py` is the Python file and `kb` is the name of the `KnowledgeBase` variable. The command connects to the Knowledge Engine, registers the KB, runs the handling loop, and on `SIGINT` or `SIGTERM` unregisters and closes cleanly.
+
 ## Examples
 
 The [`examples/`](./examples/) directory contains runnable examples covering all major features. Each example has inline comments explaining the code.
@@ -65,6 +75,7 @@ The [`examples/`](./examples/) directory contains runnable examples covering all
 | [07-testing/](./examples/07-testing/) | Writing tests with the in-memory `TestClient` |
 | [08-async_handlers.py](./examples/08-async_handlers.py) | Async and sync REACT handlers side by side |
 | [09-sparql-store/](./examples/09-sparql-store/) | Connecting a SPARQL store as a knowledge base |
+| [10-cli.py](./examples/10-cli.py) | Start a KB via the `knowledge-mapper run` CLI |
 
 See the [examples README](./examples/README.md) for prerequisites and setup instructions.
 

examples/10-cli.py

@@ -0,0 +1,40 @@
+"""CLI example: define a KnowledgeBase and let ``knowledge-mapper run`` start it.
+
+This file deliberately contains no ``asyncio.run`` boilerplate or
+``if __name__ == "__main__"`` block. The CLI takes care of the full lifecycle —
+connect, register, run the handling loop, and unregister on Ctrl+C.
+
+Run with:
+
+    knowledge-mapper run 10-cli.py:kb
+
+(from the ``examples/`` directory, with the package installed). Press Ctrl+C
+to stop; the CLI handles SIGINT/SIGTERM and shuts the KB down cleanly.
+"""
+
+from shared import get_example_logger
+
+from knowledge_mapper import KnowledgeBase
+
+EXAMPLE_NAME = "cli"
+logger = get_example_logger(EXAMPLE_NAME)
+
+kb = KnowledgeBase(
+    id="http://example.org/knowledge-mapper/cli#kb",
+    name="example-cli-kb",
+    description="A KB started via the knowledge-mapper CLI.",
+    ke_url="http://localhost:8280/rest",
+)
+
+
+@kb.answer_ki(
+    name="example-answer-ki",
+    graph_pattern="""
+        ?question a ex:Question .
+        ?question ex:hasText ?text .
+    """,
+    prefixes={"ex": "http://example.org/knowledge-mapper/cli#"},
+)
+def example_answer_ki(binding_set, info):
+    logger.info("Handling a call to the example answer KI.")
+    return binding_set

examples/README.md

@@ -15,6 +15,7 @@ The best way to get started with the Knowledge Mapper is by exploring the exampl
 | **07-testing/** | Demonstrates how to write tests for your knowledge base using the fake client |
 | **08-async_handlers.py** | Demonstrates async REACT handlers and how they differ from sync handlers |
 | **09-sparql-store/** | Connecting a SPARQL store as a knowledge base |
+| **10-cli.py** | Start a KB via the `knowledge-mapper run` CLI — no `asyncio.run` boilerplate |
 
 ## Prerequisites
 
@@ -61,6 +62,17 @@ python 01-basic.py
 
 Most examples will start the knowledge mapper and connect to the Knowledge Engine. Press `Ctrl+C` to stop.
 
+### Running via the CLI
+
+Example **10-cli.py** is started through the `knowledge-mapper` CLI instead of `python`:
+
+```bash
+cd examples
+knowledge-mapper run 10-cli.py:kb
+```
+
+The CLI handles `connect`, `register`, the handling loop, and graceful shutdown on `Ctrl+C`.
+
 ### Running Tests
 
 The `07-testing/` example includes test examples. Run them with:

pyproject.toml

@@ -9,8 +9,12 @@ dependencies = [
     "pydantic>=2.12.5",
     "pydantic-settings[yaml]>=2.13.1",
     "rdflib>=7.6.0",
+    "typer>=0.15",
 ]
 
+[project.scripts]
+knowledge-mapper = "knowledge_mapper.cli:app"
+
 [dependency-groups]
 dev = [
     "pytest>=9.0.2",

src/knowledge_mapper/cli.py

@@ -0,0 +1,94 @@
+"""CLI for the Knowledge Mapper. See ``knowledge-mapper --help``."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import importlib.util
+import signal
+import sys
+from pathlib import Path
+from typing import Any
+
+import typer
+
+app = typer.Typer(
+    help="Knowledge Mapper CLI.",
+    no_args_is_help=True,
+)
+
+
+@app.callback()
+def _main() -> None:
+    """Knowledge Mapper CLI."""
+
+
+def load_kb(spec: str) -> Any:
+    """Load a knowledge base instance from a ``path/to/file.py:attr`` spec.
+
+    The file's parent directory is added to ``sys.path`` so the loaded module
+    can import sibling modules just like when run directly with ``python``.
+    """
+    if ":" not in spec:
+        raise ValueError(f"Invalid spec {spec!r}: expected 'path/to/file.py:attr'.")
+    path_part, attr = spec.split(":", 1)
+    file = Path(path_part).resolve()
+
+    parent = str(file.parent)
+    if parent not in sys.path:
+        sys.path.insert(0, parent)
+
+    module_spec = importlib.util.spec_from_file_location(file.stem, file)
+    assert module_spec is not None and module_spec.loader is not None
+    module = importlib.util.module_from_spec(module_spec)
+    module_spec.loader.exec_module(module)
+    return getattr(module, attr)
+
+
+_SHUTDOWN_SIGNALS = (signal.SIGINT, signal.SIGTERM)
+
+
+async def run_kb(kb: Any) -> None:
+    """Run the full lifecycle of a knowledge base.
+
+    Connects, registers, and starts the handling loop. On SIGINT/SIGTERM the
+    loop is cancelled and the KB is unregistered and closed before returning.
+    """
+    await kb.connect()
+    await kb.register()
+
+    loop = asyncio.get_running_loop()
+    handling_task = asyncio.create_task(kb.start_handling_loop())
+
+    def _request_shutdown() -> None:
+        handling_task.cancel()
+
+    installed: list[int] = []
+    for sig in _SHUTDOWN_SIGNALS:
+        try:
+            loop.add_signal_handler(sig, _request_shutdown)
+            installed.append(sig)
+        except NotImplementedError:
+            pass
+
+    try:
+        with contextlib.suppress(asyncio.CancelledError):
+            await handling_task
+    finally:
+        for sig in installed:
+            loop.remove_signal_handler(sig)
+        await kb.unregister()
+        await kb.close()
+
+
+@app.command("run")
+def run_command(
+    spec: str = typer.Argument(
+        ...,
+        metavar="PATH:ATTR",
+        help="Python file and attribute, e.g. 'my_app.py:kb'.",
+    ),
+) -> None:
+    """Run a KnowledgeBase defined in a Python file."""
+    kb = load_kb(spec)
+    asyncio.run(run_kb(kb))

tests/test_cli.py

@@ -0,0 +1,177 @@
+"""Tests for the ``knowledge-mapper`` CLI."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from knowledge_mapper.cli import load_kb, run_kb
+
+
+class _FakeKB:
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+
+    async def connect(self) -> None:
+        self.calls.append("connect")
+
+    async def register(self) -> None:
+        self.calls.append("register")
+
+    async def start_handling_loop(self) -> None:
+        self.calls.append("start_handling_loop")
+
+    async def unregister(self) -> None:
+        self.calls.append("unregister")
+
+    async def close(self) -> None:
+        self.calls.append("close")
+
+
+def _write_module(tmp_path: Path, body: str) -> Path:
+    file = tmp_path / "user_app.py"
+    file.write_text(body)
+    return file
+
+
+def test_load_kb_returns_named_attribute_from_file(tmp_path: Path):
+    file = _write_module(
+        tmp_path,
+        "class FakeKB:\n    pass\n\nkb = FakeKB()\n",
+    )
+
+    result = load_kb(f"{file}:kb")
+
+    assert type(result).__name__ == "FakeKB"
+
+
+def test_load_kb_rejects_spec_without_colon():
+    with pytest.raises(ValueError, match="path/to/file.py:attr"):
+        load_kb("no_colon_here.py")
+
+
+def test_load_kb_raises_when_file_missing(tmp_path: Path):
+    missing = tmp_path / "no_such_file.py"
+
+    with pytest.raises(FileNotFoundError, match=str(missing)):
+        load_kb(f"{missing}:kb")
+
+
+def test_load_kb_raises_when_attribute_missing(tmp_path: Path):
+    file = _write_module(tmp_path, "other = 42\n")
+
+    with pytest.raises(AttributeError, match="kb"):
+        load_kb(f"{file}:kb")
+
+
+async def test_run_kb_invokes_full_lifecycle_in_order():
+    kb = _FakeKB()
+
+    await run_kb(kb)
+
+    assert kb.calls == [
+        "connect",
+        "register",
+        "start_handling_loop",
+        "unregister",
+        "close",
+    ]
+
+
+async def test_run_kb_cleans_up_when_handling_loop_raises():
+    class CrashingKB(_FakeKB):
+        async def start_handling_loop(self) -> None:
+            self.calls.append("start_handling_loop")
+            raise RuntimeError("boom")
+
+    kb = CrashingKB()
+
+    with pytest.raises(RuntimeError, match="boom"):
+        await run_kb(kb)
+
+    assert kb.calls == [
+        "connect",
+        "register",
+        "start_handling_loop",
+        "unregister",
+        "close",
+    ]
+
+
+async def test_run_kb_shuts_down_gracefully_on_sigint():
+    import asyncio
+    import os
+    import signal
+
+    class HangingKB(_FakeKB):
+        async def start_handling_loop(self) -> None:
+            self.calls.append("start_handling_loop")
+            asyncio.get_running_loop().call_later(
+                0.05, lambda: os.kill(os.getpid(), signal.SIGINT)
+            )
+            await asyncio.Event().wait()
+
+    kb = HangingKB()
+    await run_kb(kb)
+
+    assert kb.calls == [
+        "connect",
+        "register",
+        "start_handling_loop",
+        "unregister",
+        "close",
+    ]
+
+
+def test_run_subcommand_executes_full_lifecycle(tmp_path: Path):
+    from typer.testing import CliRunner
+
+    from knowledge_mapper.cli import app
+
+    log = tmp_path / "calls.log"
+    kb_file = _write_module(
+        tmp_path,
+        f"""
+from pathlib import Path
+
+LOG = Path({str(log)!r})
+
+class FakeKB:
+    async def connect(self):
+        with LOG.open('a') as f: f.write('connect\\n')
+    async def register(self):
+        with LOG.open('a') as f: f.write('register\\n')
+    async def start_handling_loop(self):
+        with LOG.open('a') as f: f.write('start\\n')
+    async def unregister(self):
+        with LOG.open('a') as f: f.write('unregister\\n')
+    async def close(self):
+        with LOG.open('a') as f: f.write('close\\n')
+
+kb = FakeKB()
+""",
+    )
+
+    result = CliRunner().invoke(app, ["run", f"{kb_file}:kb"])
+
+    assert result.exit_code == 0, result.output
+    assert log.read_text().splitlines() == [
+        "connect",
+        "register",
+        "start",
+        "unregister",
+        "close",
+    ]
+
+
+def test_load_kb_allows_sibling_imports(tmp_path: Path):
+    (tmp_path / "sibling.py").write_text("MESSAGE = 'hello'\n")
+    file = _write_module(
+        tmp_path,
+        "from sibling import MESSAGE\n\nkb = MESSAGE\n",
+    )
+
+    result = load_kb(f"{file}:kb")
+
+    assert result == "hello"