Implementation — Context Nexus

Technical architecture

Five components, one coherent system.

Document storage

Registers as Git-versioned files. The source of truth. Simple, auditable, diffable, compatible with all existing CI/CD workflows.

Vector index

Knowledge artifacts split into chunks, embedded, and indexed for semantic retrieval. The engine of RAG mode.

Agent runtime

The agent execution engine. Manages the plan–tool–observation loop, session memory, and MCP Skill calls.

Context Assembler

The orchestration layer specific to Context Nexus. Traverses the inheritance chain, validates Contracts, assembles the task brief, and passes it to the agent runtime.

Observability

Traceability of agent decisions and outputs. Essential for detecting quality drift and calibrating the registers.

Recommended stack — POC

Local development

Component	Tool	Register(s)
Document storage	`Git / filesystem`	Knowledge · Intent · Contracts · Operations
Vector index	`ChromaDB (local)`	Knowledge
Agent runtime	`LangChain / LangGraph`	All
Context Assembler	`Python function`	All
MCP Skills	`mcp Python SDK`	Contracts · Operations
Observability	`Langfuse (local)`	All

Recommended stack — Production

Production-grade

Component	Recommended tool	Notes
Document storage	`Git + CI/CD pipeline`	Validation and indexing on each PR
Vector index	`Weaviate / Pinecone`	Schema and metadata management
Agent runtime	`LangGraph + Redis`	Distributed session memory
Context Assembler	`FastAPI service`	Stateless, horizontally scalable
MCP Skills	`MCP server cluster`	Shared organizational registry
Observability	`Langfuse Cloud`	3 traceability dimensions

Context Assembler

The central component.

The Context Assembler takes a task as input and produces a structured task brief as output. This brief is all the agent needs to work.

from dataclasses import dataclass, field
from typing import Any

@dataclass
class TaskBrief:
    task_id: str
    system_prompt: str          # Knowledge conventions + Intent directives + Contracts assertions
    context_injection: str      # Current spec + applicable Decision Directives
    rag_index: str              # Chroma collection name for RAG calls
    mcp_skills: list[str]       # Callable MCP skill names for this task
    metadata: dict[str, Any] = field(default_factory=dict)

class ContextAssembler:
    def __init__(self, nexus_root: str, rag_client, langfuse):
        self.nexus_root = nexus_root
        self.rag = rag_client
        self.lf = langfuse

    def assemble(self, task_id: str, team: str, spec_path: str) -> TaskBrief:
        with self.lf.trace(name="assemble", input={"task_id": task_id}) as trace:
            # 1. Traverse inheritance chain: org → intermediate → team
            contracts = self._load_contracts(team, trace)
            directives = self._load_directives(team, trace)
            conventions = self._load_conventions(team, trace)

            # 2. Build system prompt (always-active, low-volume, high-density)
            system_prompt = "\n\n".join([conventions, directives, contracts])

            # 3. Load spec for context injection
            spec = open(f"{self.nexus_root}/{spec_path}").read()

            # 4. Resolve available MCP skills for this team
            skills = self._resolve_skills(team)

            brief = TaskBrief(
                task_id=task_id,
                system_prompt=system_prompt,
                context_injection=spec,
                rag_index=f"knowledge-{team}",
                mcp_skills=skills,
            )
            trace.update(output={"brief_size": len(system_prompt)})
            return brief

    def _load_contracts(self, team: str, trace) -> str:
        # Traverse Org → Intermediate → Team, block unapproved exceptions
        contracts = []
        for level_contracts in self._traverse_inheritance("contracts", team):
            for artifact in level_contracts:
                if artifact.get("exception-to") and not artifact.get("exception-approved-by"):
                    trace.event(name="blocked_contract", input={"path": artifact["path"]})
                    continue  # block unapproved exception — never silently include
                contracts.append(artifact["body"])
        return "\n\n".join(contracts)

Knowledge indexing (RAG)

Knowledge artifacts are chunked by atomic concept, then embedded into a vector index. See the Knowledge register →

from langchain.text_splitter import MarkdownHeaderTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import SentenceTransformerEmbeddings

def index_knowledge(nexus_root: str, team: str) -> Chroma:
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=[("##", "section"), ("###", "subsection")]
    )
    docs = []
    for level in ["org", f"intermediate/{team_bu}", f"teams/{team}"]:
        for path in glob(f"{nexus_root}/{level}/knowledge/**/*.md"):
            chunks = splitter.split_text(open(path).read())
            for chunk in chunks:
                chunk.metadata.update({"level": level, "source": path})
            docs.extend(chunks)

    return Chroma.from_documents(
        docs,
        SentenceTransformerEmbeddings(model_name="all-MiniLM-L6-v2"),
        persist_directory=f".chroma/{team}"
    )

Operations → MCP Skills

Operations runbooks are encapsulated as callable tools. See the Operations register →

from mcp.server import MCPServer
from skills.catalog_indexation_lag import catalog_indexation_lag
from skills.performance_test import performance_test
from skills.mutation_score import mutation_score

server = MCPServer(name="context-nexus-skills")

# Operations runbooks — executable, not just readable
server.register_tool(catalog_indexation_lag)

# Contracts verification tools — return pass/fail + metrics
server.register_tool(performance_test)
server.register_tool(mutation_score)

if __name__ == "__main__":
    server.run(host="0.0.0.0", port=8765)

Observability

Three dimensions with Langfuse.

Assembly decisions

Which Contracts were loaded, which artifacts excluded (unapproved exceptions, stale artifacts), which RAG results retrieved. These traces detect context drift before it affects outputs.

Quality gate results

Each call to a Contracts verification Skill is traced with its pass/fail result and associated metrics. A rising failure rate on a specific gate signals a regression in generated code — or a poorly calibrated gate.

RAG quality

Chunks retrieved for each query are traced with their similarity score. A declining average score indicates Knowledge is no longer aligned with agent queries: artifacts may be stale or chunking needs review.

from langfuse import Langfuse

lf = Langfuse()

def traced_assembly(task_id: str, assembler: ContextAssembler, **kwargs) -> TaskBrief:
    with lf.trace(name="context-assembly", user_id=kwargs.get("user")) as trace:
        trace.span(name="inheritance-traversal")
        brief = assembler.assemble(task_id=task_id, **kwargs)
        trace.score(name="brief-size", value=len(brief.system_prompt))
        return brief

def traced_rag(query: str, collection: str, k: int = 5):
    with lf.trace(name="rag-retrieval") as trace:
        results = chroma.similarity_search_with_score(query, k=k, collection=collection)
        avg_score = sum(score for _, score in results) / len(results)
        trace.score(name="avg-similarity", value=avg_score)  # alert if declining
        return [doc for doc, _ in results]

Advanced

Multi-agent architecture.

In a multi-agent context, the Context Assembler becomes an orchestrating agent. It receives a complex task, decomposes it, creates a specialized task brief for each sub-task, and delegates to worker agents. Each worker operates with its own isolated context, exactly Deep Agents mode: sub-agents see only final results, not intermediate steps of others.

Bootstrap

Start in 2 hours.

The minimum viable setup to test Context Nexus on a real project.

File structure to create

nexus/
├── org/
│   ├── knowledge/
│   │   └── conventions.md          # register: knowledge, level: org
│   ├── intent/
│   │   └── directives/
│   │       └── security.md         # register: intent, level: org
│   └── contracts/
│       └── security.md             # register: contracts, level: org
└── teams/
    └── my-team/
        ├── knowledge/
        │   └── architecture.md     # register: knowledge, level: team
        ├── intent/
        │   ├── directives/
        │   │   └── api-patterns.md
        │   └── specs/
        │       └── TICKET-42.md    # register: intent, consumption-mode: context-injection
        ├── contracts/
        │   └── quality-gates.md    # register: contracts, level: team
        └── operations/
            └── runbooks/
                └── deploy.md       # register: operations, consumption-mode: skill

Install

pip install langchain langchain-community chromadb langfuse mcp openai

Start Langfuse locally

docker run -d -p 3000:3000 langfuse/langfuse:latest

First working agent

from context_assembler import ContextAssembler
from langchain_openai import ChatOpenAI
from langfuse import Langfuse

assembler = ContextAssembler(
    nexus_root="./nexus",
    rag_client=chroma_client,
    langfuse=Langfuse(),
)

# Assemble the task brief for ticket TICKET-42
brief = assembler.assemble(
    task_id="TICKET-42",
    team="my-team",
    spec_path="teams/my-team/intent/specs/TICKET-42.md",
)

# Run the agent with its assembled context
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke([
    {"role": "system", "content": brief.system_prompt},
    {"role": "user", "content": brief.context_injection},
])
print(response.content)

The implementation doesn't make the framework. This guide provides the plumbing. The value comes from the quality of the artifacts the team feeds into the registers.

Related registers

Knowledge Intent Contracts Operations

Technical Guide