Knowledge Mapper

The Knowledge Mapper is a Python SDK for connecting your applications to the TNO Knowledge Engine (TKE) network. Define knowledge interactions with decorators, use typed binding models, and let the SDK handle registration, polling, and data exchange with the network.

The best way to learn to work with the Knowledge Mapper is to check out the examples. Users that are familiar with Python's FastAPI will recognize many concepts from that package.

Quick Start

pip install knowledge_mapper

import asyncio
from knowledge_mapper import KnowledgeBase

kb = KnowledgeBase(
    id="http://example.org/my-kb",
    name="my-kb",
    description="A simple example KB.",
    ke_url="http://localhost:8280/rest",
)


@kb.answer_ki(
    name="greeting",
    graph_pattern="?s <http://example.org/says> ?message .",
)
def handle(binding_set, info):
    return binding_set


async def main():
    await kb.connect()
    await kb.register()
    await kb.start_handling_loop()


asyncio.run(main())

Installation

Install from PyPI:

pip install knowledge_mapper

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:

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/ directory contains runnable examples covering all major features. Each example has inline comments explaining the code.

Example	Description
01-basic.py	Minimal knowledge base with a simple ANSWER knowledge interaction
02-binding_models.py	Pydantic-style binding models for type-safe bindings
03-ask_interaction.py	Outgoing ASK knowledge interactions with typed binding models
04-post_measurement.py	POST interactions for pushing measurements or data to the network
05-custom-settings/	Custom settings subclass with YAML configuration
06-dependency_injection.py	Dependency injection for handlers (database connections, configs, etc.)
07-testing/	Writing tests with the in-memory TestClient
08-async_handlers.py	Async and sync REACT handlers side by side
09-sparql-store/	Connecting a SPARQL store as a knowledge base
10-cli.py	Start a KB via the knowledge-mapper run CLI
11-custom_datatypes.py	Literals with custom (non-XSD) datatypes in binding models
12-dockerized/	Package a KB as a Docker image and run it with a Smart Connector via Docker Compose

See the examples README for prerequisites and setup instructions.

API Reference

KnowledgeBase

The main entry point. Create one, register knowledge interactions with decorators, then connect and start handling.

from knowledge_mapper import KnowledgeBase

kb = KnowledgeBase(
    id="http://example.org/my-kb",
    name="my-kb",
    description="...",
    ke_url="http://localhost:8280/rest",
)

Lifecycle:

await kb.connect()  # Verify the Smart Connector is reachable
await kb.register()  # Register KB and all KIs with the Smart Connector
await kb.start_handling_loop()  # Start concurrent long-polling for incoming requests
await kb.unregister()  # Unregister from the Smart Connector
await kb.close()  # Close the HTTP client

Registering Knowledge Interactions (decorators):

# ANSWER — respond to incoming queries
@kb.answer_ki(name="...", graph_pattern="...", prefixes={...})
def my_handler(binding_set, info):
    return binding_set


# REACT — handle incoming POST data
@kb.react_ki(
    name="...", argument_graph_pattern="...", result_graph_pattern="...", prefixes={...}
)
def my_react_handler(binding_set, info):
    return result_binding_set

Registering Knowledge Interactions (non-decorator):

# ASK — query the network (no handler needed)
kb.ask_ki(name="...", graph_pattern="...", binding_model=MyModel, prefixes={...})

# POST — push data to the network (no handler needed)
kb.post_ki(
    name="...", argument_graph_pattern="...", result_graph_pattern="...", prefixes={...}
)

Outgoing interactions:

result = await kb.ask(binding_set, ki_name="...")
result = await kb.post(binding_set, ki_name="...")

If the KI has a binding model, pass it to get a typed result (no cast() needed). It must be the model the KI was registered with, otherwise a ValueError is raised:

# Sequence[MyModel] / Sequence[ResultModel]
people = await kb.ask(binding_set, ki_name="...", binding_model=MyModel)
results = await kb.post(binding_set, ki_name="...", result_binding_model=ResultModel)

To also get the exchange info reported by the Knowledge Engine (which knowledge interactions were involved, with status, timing and the bindings they exchanged), use the *_with_info variants. They return an ExchangeResult with binding_set and exchange_info, a list of AskExchangeInfo (exchanges with ANSWER KIs) and PostExchangeInfo (exchanges with REACT KIs). Both kinds can occur for either ASK or POST, because the reasoner may combine KIs:

from knowledge_mapper import AskExchangeInfo, PostExchangeInfo

result = await kb.ask_with_info(binding_set, ki_name="...", binding_model=MyModel)
for info in result.exchange_info:
    print(info.knowledge_base_id, info.status, info.failed_message)
    match info:
        case AskExchangeInfo():
            print("answered:", info.binding_set)
        case PostExchangeInfo():
            print("reacted:", info.argument_binding_set, info.result_binding_set)

The bindings in the exchange info are raw (dict[str, str]), because they follow the other KI's graph pattern rather than your binding model.

Sync handlers can use ask_sync() / post_sync() and ask_with_info_sync() / post_with_info_sync().

BindingModel

A Pydantic BaseModel subclass that maps Python types to RDF N3 encoding automatically.

from knowledge_mapper import BindingModel, Uri, Literal


class PersonBinding(BindingModel):
    person: Uri  # URIRef, serialized as <...>
    name: Literal[str]  # Python str, serialized as "..."^^xsd:string
    age: Literal[int]  # Python int, serialized as "..."^^xsd:integer

Use BindingModel for type safety and automatic serialization. Use raw BindingSet (Sequence[dict[str, str]]) for passthrough data.

Literal[T] infers an XSD datatype from the Python value. For custom (non-XSD) datatypes:

from typing import Annotated
from knowledge_mapper import BindingModel, Datatype, Literal, RdfLiteral


class ObservationBinding(BindingModel):
    # Python float, serialized as "..."^^ex:celsius; other incoming datatypes are rejected
    temperature: Annotated[Literal[float], Datatype("http://example.org/celsius")]
    # rdflib.Literal kept as-is; any datatype or language tag is preserved
    remark: RdfLiteral

Depends — Dependency Injection

Handlers can declare dependencies using Depends() in Annotated type hints. The framework resolves them at call time.

from typing import Annotated
from knowledge_mapper import Depends


def get_db() -> MyDatabase:
    return MyDatabase(url="...")


@kb.answer_ki(name="...", graph_pattern="...")
def handler(
    binding_set: list[PersonBinding],
    info: KnowledgeInteractionInfo,
    db: Annotated[MyDatabase, Depends(get_db)],
) -> list[PersonBinding]:
    return db.query(binding_set)

Dependencies can be sync or async, support nesting, and are cached per invocation by default (cache=True).

Requesting knowledge base: to know which KB initiated an incoming call (e.g. for authorization), declare a parameter named requesting_kb_id on a handler or dependency factory, or use the built-in get_requesting_kb_id factory:

from knowledge_mapper import KnowledgeBaseId, get_requesting_kb_id


def authorize(requesting_kb_id: KnowledgeBaseId) -> KnowledgeBaseId:
    if requesting_kb_id not in ALLOWED_KBS:
        raise PermissionError(requesting_kb_id)
    return requesting_kb_id


@kb.answer_ki(name="...", graph_pattern="...")
def handler(
    binding_set: BindingSet,
    info: KnowledgeInteraction,
    requesting_kb_id: KnowledgeBaseId,  # injected by name
    caller: Annotated[KnowledgeBaseId, Depends(get_requesting_kb_id)],  # same value
    authorized: Annotated[KnowledgeBaseId, Depends(authorize)],
) -> BindingSet: ...

In tests, pass it explicitly: await kb.call(binding_set, "ki-name", requesting_kb_id=...).

KnowledgeBaseBuilder

Build a KnowledgeBase from settings. Returned by KnowledgeBase.from_settings().

from knowledge_mapper import KnowledgeBase, KnowledgeBaseSettings

settings = KnowledgeBaseSettings(...)
builder = KnowledgeBase.from_settings(settings)

# Attach handlers for ANSWER/REACT KIs defined in settings
builder.handler("my-answer-ki", my_handler_func)

# ASK/POST KIs are auto-registered from settings
kb = builder.build()

Configuration

KnowledgeBaseSettings is a Pydantic BaseSettings subclass that supports configuration from (highest priority first):

1. Keyword arguments
2. Environment variables (delimiter __, e.g. KNOWLEDGE_BASE__ID)
3. YAML config file
4. JSON config file
5. Field defaults

Example YAML configuration:

knowledge_base:
  id: "http://example.org/my-kb"
  name: "my-kb"
  description: "My knowledge base"
knowledge_engine_endpoint: "http://localhost:8280/rest"
knowledge_interactions:
  - name: my-answer-ki
    type: AnswerKnowledgeInteraction
    prefixes:
      ex: "http://example.org/"
    graph_pattern: "?s ?p ?o ."

Subclass KnowledgeBaseSettings to add application-specific settings:

from knowledge_mapper import KnowledgeBaseSettings


class AppSettings(KnowledgeBaseSettings):
    db_url: str = "sqlite:///./app.db"

Development

uv sync              # Install dependencies
uv run pytest        # Run tests
uv run ruff check .  # Lint
uv run ruff format . # Format

Tests use TestClient, an in-memory fake Smart Connector — no live KE runtime needed for unit tests. For integration tests, start a KE runtime with Docker:

docker compose -f examples/compose.yaml up -d
uv run pytest
docker compose -f examples/compose.yaml down

See CONTRIBUTING.md for detailed development and contribution guidelines.