Implementation Scaffold: Indigenous AI DSL Server

This document provides concrete code patterns, configuration templates, and implementation guidance for the cloud agent building the server.

1. Project Setup

pyproject.toml

```toml [build-system] requires = ["hatchling"] build-backend = "hatchling.build"

[project] name = "indigenous-ai-dsl-server" version = "1.0.0" description = "Dual-protocol semantic framework server (LSP + MCP) for the Indigenous AI Collaborative Platform" requires-python = ">=3.10" license = { text = "Indigenous Knowledge Stewardship License" } authors = [ { name = "IAIP Community" } ]

dependencies = [ "pygls>=1.3.0", "mcp>=1.0.0", "pydantic>=2.0", "pyyaml>=6.0", "networkx>=3.0", ]

[project.optional-dependencies] ontology = [ "rdflib>=7.0", ] dev = [ "pytest>=7.0", "pytest-asyncio>=0.21", "ruff>=0.1", ]

[project.scripts] iaip-dsl = "indigenous_ai_dsl_server.main:main"

[tool.ruff] line-length = 100 ```

main.py

```python """Entry point for the Indigenous AI DSL Server.""" import argparse import asyncio from .server import IAIPDSLServer

def main(): parser = argparse.ArgumentParser(description="Indigenous AI DSL Server") parser.add_argument("--mode", choices=["lsp", "mcp", "dual"], default="dual", help="Protocol mode: lsp (IDE), mcp (agents), dual (both)") parser.add_argument("--registry", default=None, help="Path to concept-registry.yaml") parser.add_argument("--wisdom-db", default=None, help="Path to wisdom.db SQLite database") args = parser.parse_args()

server = IAIPDSLServer(
    registry_path=args.registry,
    wisdom_db_path=args.wisdom_db,
)

if args.mode == "lsp":
    server.start_lsp()
elif args.mode == "mcp":
    asyncio.run(server.start_mcp())
else:
    asyncio.run(server.start_dual())

if name == "main": main() ```

2. Core Data Model

registry/concept.py

```python """Concept data model aligned with medicine-wheel-ontology-core.""" from enum import Enum from typing import Optional from pydantic import BaseModel, Field

class Direction(str, Enum): EAST = "east" SOUTH = "south" WEST = "west" NORTH = "north" CENTER = "center"

class NodeType(str, Enum): CONCEPT = "concept" PRACTICE = "practice" KNOWLEDGE_SYSTEM = "knowledge_system" FEELING = "feeling" PRINCIPLE = "principle" CEREMONY = "ceremony"

class RelationType(str, Enum): FOUNDATIONAL = "foundational" COMPLEMENTARY = "complementary" ENABLES = "enables" CONTEXTUALIZES = "contextualizes" CEREMONIAL = "ceremonial" STEWARDS = "stewards" SERVES = "serves" GIVES_BACK_TO = "gives_back_to" BORN_FROM = "born_from" KINSHIP_OF = "kinship_of" APPLIES = "applies" EXTENDS = "extends" GROUNDS = "grounds" GATHERS = "gathers" DOCUMENTS = "documents" TRACKS = "tracks" MANIFESTS = "manifests"

class OcapFlags(BaseModel): ownership: str = "community" control: str = "steward" access: str = "open" # open | restricted | ceremony_required possession: str = "shared"

class IndigenousConnections(BaseModel): two_eyed_seeing: Optional[str] = Field(None, alias="twoEyedSeeing") polycentric: Optional[str] = None relational: Optional[str] = None ceremonial: Optional[str] = None

class Config:
    populate_by_name = True

class TypedRelation(BaseModel): source_id: str target_id: str relation_type: RelationType direction: Optional[Direction] = None weight: float = 1.0 narrative_context: Optional[str] = None

class ConceptNode(BaseModel): """Core concept model — aligned with RelationalNode from medicine-wheel-ontology-core.""" id: str name: str node_type: NodeType = NodeType.CONCEPT category: str definition: str direction: Direction = Direction.CENTER

related_concepts: list[str] = Field(default_factory=list, alias="relatedConcepts")
relationships: list[TypedRelation] = Field(default_factory=list)

indigenous_connections: IndigenousConnections = Field(
    default_factory=IndigenousConnections,
    alias="indigenousConnections"
)

ocap: OcapFlags = Field(default_factory=OcapFlags)

sources: list[str] = Field(default_factory=list)
usage_contexts: list[str] = Field(default_factory=list, alias="usageContext")
examples: list[str] = Field(default_factory=list)
notes: Optional[str] = None

ceremony_born: Optional[str] = None
wilson_alignment: Optional[float] = None

class Config:
    populate_by_name = True

```

3. Registry Loader

registry/loader.py

```python """Load concept registry from YAML, build in-memory index.""" import yaml from pathlib import Path from .concept import ConceptNode, Direction, NodeType

class ConceptRegistry: def init(self): self._concepts: dict[str, ConceptNode] = {} self._name_index: dict[str, str] = {} # lowercase name → id self._category_index: dict[str, list[str]] = {} self._direction_index: dict[str, list[str]] = {}

def load_yaml(self, path: Path):
    """Load concepts from YAML file."""
    with open(path) as f:
        data = yaml.safe_load(f)

    for key, entry in data.get("concepts", {}).items():
        concept = self._parse_concept(entry)
        self._index_concept(concept)

def _parse_concept(self, entry: dict) -> ConceptNode:
    """Parse YAML entry into ConceptNode, inferring missing fields."""
    # Infer direction from category if not specified
    if "direction" not in entry:
        entry["direction"] = self._infer_direction(entry.get("category", ""))

    # Infer node_type from category
    if "node_type" not in entry:
        entry["node_type"] = self._infer_node_type(entry.get("category", ""))

    return ConceptNode(**entry)

def _index_concept(self, concept: ConceptNode):
    self._concepts[concept.id] = concept
    self._name_index[concept.name.lower()] = concept.id

    cat = concept.category
    self._category_index.setdefault(cat, []).append(concept.id)

    d = concept.direction.value
    self._direction_index.setdefault(d, []).append(concept.id)

def lookup(self, query: str) -> ConceptNode | None:
    """Look up concept by ID or name."""
    if query in self._concepts:
        return self._concepts[query]
    return self._concepts.get(self._name_index.get(query.lower()))

def search(self, keyword: str) -> list[ConceptNode]:
    """Search concepts by keyword in name, definition, or category."""
    keyword = keyword.lower()
    return [c for c in self._concepts.values()
            if keyword in c.name.lower()
            or keyword in c.definition.lower()
            or keyword in c.category.lower()]

def by_direction(self, direction: str) -> list[ConceptNode]:
    ids = self._direction_index.get(direction, [])
    return [self._concepts[i] for i in ids]

def by_category(self, category: str) -> list[ConceptNode]:
    ids = self._category_index.get(category, [])
    return [self._concepts[i] for i in ids]

def all_concepts(self) -> list[ConceptNode]:
    return list(self._concepts.values())

@staticmethod
def _infer_direction(category: str) -> str:
    mapping = {
        "Framework": "south",
        "Feeling/Capacity": "west",
        "Emotional/Developmental": "west",
        "Knowledge System": "east",
        "Integration": "center",
    }
    return mapping.get(category, "center")

@staticmethod
def _infer_node_type(category: str) -> str:
    mapping = {
        "Framework": "concept",
        "Feeling/Capacity": "feeling",
        "Emotional/Developmental": "feeling",
        "Knowledge System": "knowledge_system",
        "Integration": "principle",
    }
    return mapping.get(category, "concept")

```

4. Relational Graph

graph/engine.py

```python """NetworkX-based relational graph engine.""" import networkx as nx from ..registry.concept import ConceptNode, TypedRelation, RelationType

class RelationalGraphEngine: def init(self): self.graph = nx.DiGraph()

def build_from_registry(self, concepts: list[ConceptNode]):
    """Build graph from concept registry."""
    for concept in concepts:
        self.graph.add_node(concept.id, **{
            "name": concept.name,
            "category": concept.category,
            "direction": concept.direction.value,
        })

    for concept in concepts:
        for related_id in concept.related_concepts:
            if not self.graph.has_edge(concept.id, related_id):
                self.graph.add_edge(concept.id, related_id,
                                    relation_type="related")

        for rel in concept.relationships:
            self.graph.add_edge(rel.source_id, rel.target_id,
                                relation_type=rel.relation_type.value,
                                weight=rel.weight,
                                narrative_context=rel.narrative_context)

def find_path(self, source: str, target: str, max_length: int = 3) -> list[str] | None:
    try:
        return nx.shortest_path(self.graph, source, target)[:max_length + 1]
    except (nx.NetworkXNoPath, nx.NodeNotFound):
        return None

def get_context(self, concept_id: str, hops: int = 1) -> dict:
    """Get all concepts within N hops."""
    if concept_id not in self.graph:
        return {"center": concept_id, "connected": []}

    connected = []
    for node in nx.single_source_shortest_path_length(self.graph, concept_id, cutoff=hops):
        if node != concept_id:
            edge_data = self.graph.get_edge_data(concept_id, node, default={})
            connected.append({
                "id": node,
                "name": self.graph.nodes[node].get("name", node),
                "relation_type": edge_data.get("relation_type", "related"),
                "distance": nx.shortest_path_length(self.graph, concept_id, node),
            })

    return {"center": concept_id, "connected": connected}

def get_relationship(self, source: str, target: str) -> dict | None:
    if self.graph.has_edge(source, target):
        return self.graph.get_edge_data(source, target)
    if self.graph.has_edge(target, source):
        data = self.graph.get_edge_data(target, source)
        data["reversed"] = True
        return data
    return None

```

5. OCAP Gate

knowledge/ocap.py

```python """OCAP governance enforcement.""" from enum import Enum from ..registry.concept import ConceptNode, OcapFlags

class AccessResult(str, Enum): GRANTED = "granted" PARTIAL = "partial" # Definition only, no Indigenous connections CEREMONY_NEEDED = "ceremony_needed"

class RequesterContext: def init(self, agent_id: str = "anonymous", is_authorized_steward: bool = False, ceremony_session_active: bool = False): self.agent_id = agent_id self.is_authorized_steward = is_authorized_steward self.ceremony_session_active = ceremony_session_active

class OcapGate: def check_access(self, concept: ConceptNode, requester: RequesterContext) -> AccessResult: access = concept.ocap.access

    if access == "open":
        return AccessResult.GRANTED

    if access == "restricted":
        if requester.is_authorized_steward:
            return AccessResult.GRANTED
        return AccessResult.PARTIAL

    if access == "ceremony_required":
        if requester.ceremony_session_active:
            return AccessResult.GRANTED
        return AccessResult.CEREMONY_NEEDED

    return AccessResult.GRANTED

def filter_response(self, concept: ConceptNode,
                    access_result: AccessResult) -> dict:
    """Filter concept data based on access result."""
    base = {
        "id": concept.id,
        "name": concept.name,
        "definition": concept.definition,
        "category": concept.category,
        "direction": concept.direction.value,
    }

    if access_result == AccessResult.GRANTED:
        base["indigenous_connections"] = concept.indigenous_connections.model_dump()
        base["related_concepts"] = concept.related_concepts
        base["usage_contexts"] = concept.usage_contexts
        base["examples"] = concept.examples
        base["ocap"] = concept.ocap.model_dump()
        base["sources"] = concept.sources

    elif access_result == AccessResult.PARTIAL:
        base["access_note"] = ("Indigenous knowledge connections for this concept "
                               "require steward authorization.")

    elif access_result == AccessResult.CEREMONY_NEEDED:
        base = {
            "id": concept.id,
            "name": concept.name,
            "access_note": ("This concept requires an active ceremony session "
                            "for full access. Please consult a knowledge steward."),
        }

    return base

```

6. MCP Tool Registration Pattern

mcp/tools.py

```python """MCP tool definitions for the IAIP DSL Server.""" from mcp.server import Server from mcp.types import Tool, TextContent

def register_mcp_tools(mcp: Server, registry, graph, ocap_gate, wisdom_ledger): """Register all IAIP MCP tools on the server."""

@mcp.tool()
async def iaip_concept_lookup(concept: str, depth: str = "full") -> list[TextContent]:
    """Look up an IAIP framework concept with Indigenous knowledge context."""
    node = registry.lookup(concept)
    if not node:
        return [TextContent(type="text",
                            text=f"Concept '{concept}' not found in registry.")]

    requester = RequesterContext(agent_id="mcp-client")
    access = ocap_gate.check_access(node, requester)
    data = ocap_gate.filter_response(node, access)

    if depth == "relational" and access == AccessResult.GRANTED:
        data["context"] = graph.get_context(node.id, hops=2)

    # Log to wisdom ledger
    await wisdom_ledger.log_usage(
        concept_id=node.id,
        event_type="lookup",
        agent_id="mcp-client",
    )

    return [TextContent(type="text", text=format_concept(data, depth))]

@mcp.tool()
async def iaip_concept_relationship(concept1: str, concept2: str,
                                     max_path_length: int = 3) -> list[TextContent]:
    """Show how two IAIP concepts relate."""
    path = graph.find_path(concept1, concept2, max_path_length)
    if not path:
        return [TextContent(type="text",
                            text=f"No path found between '{concept1}' and '{concept2}'.")]

    rel = graph.get_relationship(concept1, concept2)
    return [TextContent(type="text",
                        text=format_relationship(concept1, concept2, path, rel))]

@mcp.tool()
async def iaip_validate_terminology(text: str,
                                     strictness: str = "standard") -> list[TextContent]:
    """Validate IAIP framework terminology usage in text."""
    results = validate_text(text, registry, strictness)
    return [TextContent(type="text", text=format_validation(results))]

# ... register remaining tools similarly

```

7. Plugin Configuration

plugin.json

```json { "name": "indigenous-ai-dsl", "version": "1.0.0", "description": "Semantic framework server for the Indigenous AI Collaborative Platform — concept definitions, relational navigation, Indigenous knowledge context, and OCAP governance", "author": "IAIP Community", "type": "lsp", "marketplace": "indigenous-ai-marketplace" } ```

.lsp.json

```json { "indigenous-ai": { "command": ["python3", "-m", "indigenous_ai_dsl_server", "--mode", "lsp"], "extensionToLanguage": { ".md": "markdown", ".txt": "text", ".py": "python", ".js": "javascript", ".ts": "typescript", ".yaml": "yaml", ".yml": "yaml", ".json": "json" }, "initializationOptions": { "registryPath": "${PLUGIN_ROOT}/data/concept-registry.yaml", "extensionsPath": "${PLUGIN_ROOT}/data/concept-registry-extensions.yaml", "wisdomDbPath": "${PLUGIN_ROOT}/data/wisdom.db" } } } ```

MCP server configuration (for agents)

```json { "mcpServers": { "iaip-dsl": { "command": "python3", "args": ["-m", "indigenous_ai_dsl_server", "--mode", "mcp"], "env": { "IAIP_REGISTRY_PATH": "/path/to/concept-registry.yaml", "IAIP_EXTENSIONS_PATH": "/path/to/concept-registry-extensions.yaml" } } } } ```

8. Testing Strategy

Test Categories

``` tests/ ├── test_registry.py # Concept loading, indexing, search ├── test_graph.py # Relationship traversal, path finding ├── test_ocap.py # Access control enforcement ├── test_lsp_handlers.py # LSP operation responses ├── test_mcp_tools.py # MCP tool input/output contracts ├── test_ceremony.py # Change protocol enforcement ├── test_wisdom_ledger.py # Usage tracking, pattern detection ├── test_ontology_bridge.py # Type alignment validation └── test_integration.py # End-to-end dual-protocol scenarios ```

Key Test Patterns

```python

test_ocap.py

def test_ceremony_required_concept_without_ceremony(): """Concept with ceremony_required access should not reveal full content.""" concept = make_concept(ocap=OcapFlags(access="ceremony_required")) requester = RequesterContext(ceremony_session_active=False) result = gate.check_access(concept, requester) assert result == AccessResult.CEREMONY_NEEDED

def test_all_concepts_have_ocap(): """Every concept in the registry must have valid OCAP flags.""" for concept in registry.all_concepts(): assert concept.ocap.ownership assert concept.ocap.control assert concept.ocap.access in ["open", "restricted", "ceremony_required"] assert concept.ocap.possession

test_graph.py

def test_every_concept_has_minimum_relationships(): """Each concept must have at least 3 related concepts.""" for concept in registry.all_concepts(): assert len(concept.related_concepts) >= 3, ( f"Concept {concept.id} has only {len(concept.related_concepts)} relations" )

test_registry.py

def test_all_concepts_have_indigenous_connections(): """Every concept must have at least 2 Indigenous knowledge connections.""" for concept in registry.all_concepts(): connections = concept.indigenous_connections filled = sum(1 for v in [connections.two_eyed_seeing, connections.polycentric, connections.relational, connections.ceremonial] if v is not None) assert filled >= 2, f"Concept {concept.id} has only {filled} Indigenous connections" ```

9. Performance Requirements

10. Deployment Options

Option A: Claude Code Plugin (LSP mode)

```bash pip install indigenous-ai-dsl-server

Plugin auto-starts via .lsp.json

```

Option B: MCP Server for Agents

```bash pip install indigenous-ai-dsl-server python -m indigenous_ai_dsl_server --mode mcp

Or configure in agent's MCP config

```

Option C: Dual Protocol (recommended for development)

```bash pip install indigenous-ai-dsl-server python -m indigenous_ai_dsl_server --mode dual

LSP on stdio, MCP on SSE port 8765

```

Option D: Docker

```dockerfile FROM python:3.12-slim WORKDIR /app COPY . . RUN pip install -e . CMD ["python", "-m", "indigenous_ai_dsl_server", "--mode", "dual"] ```