Skip to main content
Noēsis doesn’t replace your agent runtime—it wraps it with observable cognition. This guide shows how to integrate your existing graphs and tools.
This guide reflects the current codebase and is updated frequently while Noēsis is in active development.

​
How adapters work

An adapter is any execution target that Noēsis can invoke.

​
The ns.solve() integration contract

ns.solve() invokes the resolved using= target in the act phase by calling invoke(), then run(), then __call__() (in that order). EpisodeRunner owns cognition, governance, and event emission. If you’re already inside an event loop, use ns.solve_async(...), which preserves the same invocation order and semantics while allowing async adapters to be awaited.

​
The execute(...) adapter protocol (advanced)

The repo also defines an Adapter.execute(...) protocol in noesis.adapters.protocols.Adapter: 
execute(*, task, episode_id, run_dir, intuition=None, seed=0, tags=None) -> Any
This protocol is used by some integrations and legacy adapters, but ns.solve() does not call execute() directly.

​
Protocol-first tool invocation contract (internal)

Noesis now includes an application-layer contract for side-effect tools in noesis.usecases.tool_invocation:
  • prepare_tool_invocation(...)
  • execute_prepared_tool_invocation(...)
This is a builder-facing integration surface for protocol adapters. The public runtime entry points (ns.run(...), ns.solve(...), ns.governed_act(...)) remain unchanged.

​
Intent

The contract separates reviewable intent from side effects:
  • prepare_tool_invocation(...) may validate, normalize, authenticate, authorize, emit candidate evidence, compute preflight bindings, and persist a PreparedToolInvocation.
  • prepare_tool_invocation(...) must not execute tool side effects.
  • execute_prepared_tool_invocation(...) loads previously prepared intent by identity and dispatches only after approval/idempotency checks.

​
Prepare -> execute workflow

Use this runbook when building a tool protocol adapter:
  1. Build ToolInvocationInput from inbound protocol data.
  2. Call prepare_tool_invocation(...) with your ports and persist the returned draft.
  3. If status is pending_approval, collect approval externally and persist ToolApprovalDecision using the same run_id + draft_id.
  4. Call execute_prepared_tool_invocation(...) with that run_id + draft_id.
  5. Return ToolExecutionResult to the caller; handle replayed/failed idempotency outcomes without re-dispatching.
Required ports:
  • prepare: ToolPayloadNormalizerPort, ToolAuthenticatorPort, ToolAuthorizerPort, ToolCandidateEmitterPort, ToolEventRecorderPort, PreparedInvocationRepositoryPort, optional ToolPreflightPort
  • execute: PreparedInvocationRepositoryPort, ApprovalDecisionRepositoryPort, IdempotencyStorePort, ToolDispatchPort, ToolEventRecorderPort
Minimal integration shape: 
from noesis.domain.tool_contract import (
    EffectKind,
    ExecutionContext,
    GovernanceContext,
    PayloadRedactionPolicy,
    RiskTier,
    SecurityContext,
    ToolIdentity,
    ToolProtocol,
)
from noesis.usecases.tool_invocation import (
    ToolInvocationInput,
    execute_prepared_tool_invocation,
    prepare_tool_invocation,
)

request = ToolInvocationInput(
    run_id="run-1",
    request_id="req-1",
    protocol=ToolProtocol.SUBPROCESS,
    tool=ToolIdentity(namespace="repo", name="write_file", version="1"),
    raw_payload={"path": "README.md", "token": "secret"},
    execution=ExecutionContext(timeout_ms=1000, retry_limit=1, idempotency_key="idem-1"),
    security=SecurityContext(
        principal_id="user:123",
        scopes=("repo:write",),
        policy_scope="repo/main",
    ),
    governance=GovernanceContext(
        effect_kind=EffectKind.WRITE,
        risk_tier=RiskTier.HIGH,
        candidate_id=None,
        requires_approval=True,
    ),
    redaction_policy=PayloadRedactionPolicy(redact_fields=("token",)),
)

prepared = prepare_tool_invocation(
    request=request,
    normalizer=normalizer,
    authenticator=authenticator,
    authorizer=authorizer,
    candidate_emitter=candidate_emitter,
    event_recorder=event_recorder,
    prepared_repository=prepared_repository,
    preflight=preflight,
)

# Later, after approval is persisted for run_id + draft_id:
result = execute_prepared_tool_invocation(
    run_id=prepared.run_id,
    draft_id=prepared.draft_id or "",
    prepared_repository=prepared_repository,
    approval_repository=approval_repository,
    idempotency_store=idempotency_store,
    dispatch=dispatch,
    event_recorder=event_recorder,
)

​
Event and identity invariants

For write + approval-required flows, prepare emits: tool.requested -> tool.validated -> tool.authn.passed -> tool.authz.passed -> action.candidate_emitted -> tool.preflight.computed -> tool.draft_created -> tool.approval.pending For approved execution, execute emits:
  • new execution: tool.approved -> tool.execution.started -> tool.execution.succeeded (or tool.execution.failed)
  • replay path: tool.approved -> tool.replayed
Identity and binding rules:
  • execute lookup key is run_id + draft_id
  • missing prepared draft raises PreparedToolInvocationNotFoundError
  • missing/non-approved decision raises ApprovalDecisionRequiredError
  • mismatched request_id, reviewed fingerprint, or impact hash raises ApprovalDecisionBindingError
  • idempotency replay or conflict returns without dispatching side effects

​
Common pitfalls

SymptomLikely causeFix
PreparedToolInvocationNotFoundErrorapproval service called execute with wrong run_id/draft_idpersist and pass durable identity unchanged across systems
ApprovalDecisionRequiredErrordecision missing or not approvedwrite an approved ToolApprovalDecision before execute
ApprovalDecisionBindingErrordecision is not bound to reviewed prepared intentstore request_id, reviewed_fingerprint, and impact_hash from the prepared artifact and verify before save
tool.replayed result with no dispatchsame idempotency key and fingerprint seen previouslytreat as successful replay; do not retry with a new side effect
summary.json / manifest.json missing while run is pausedexpected non-terminal lifecycle stateinspect latest run.state_projection links; resume and terminate run to produce final artifacts

​
Runtime bridge guardrails (current)

The runtime continuation bridge currently enforces the following:
  • only ToolProtocol.SUBPROCESS is supported for prepare/resume bridging
  • resume lookup expects exactly one pending draft for the run (load_pending_for_run)
  • multiple pending drafts for one run raise AmbiguousPreparedToolInvocationError
If a non-subprocess draft reaches runtime continuation, Noēsis raises UnsupportedToolProtocolError.

​
Subprocess payload contract

For ToolProtocol.SUBPROCESS, the normalized payload supports only:
  • argv: required, non-empty list[str]
  • cwd: optional str
  • env: optional dict[str, str]
  • timeout_ms: optional int > 0 (falls back to execution default when omitted)
Unknown payload keys fail validation before dispatch.

​
Operational runbook: inspect and resume pending drafts

Prepared/approval/idempotency records are persisted under the run directory:
  • tool_invocations/prepared/*.json
  • tool_invocations/approvals/*.json
  • tool_invocations/idempotency/*.json
Recommended flow:
  1. Confirm one pending draft exists in tool_invocations/prepared/ for the target run.
  2. Persist an approved ToolApprovalDecision bound to the same run_id + draft_id.
  3. Resume the run (ns.resume_run(...)) so continuation can execute the pending draft through the runtime bridge.

​
Runtime evidence checks (pause/continue)

When a run pauses before side effects, inspect runtime events to confirm bridge evidence is complete:
  • expect phase="runtime", event_type="run.interrupt" and run.checkpoint
  • expect a phase="runtime", event_type="run.state_projection" event
  • on paused runs, run.state_projection.payload.links should contain only events and learn (no summary/manifest yet)
  • on paused runs, run.state_projection.payload.status should match the pause status (for example interrupted)
After approval and ns.resume_run(...) completes terminally, the latest run.state_projection should include terminal links (summary, manifest) and state finalization artifacts will be present. When you call ns.solve() with using=, Noēsis:
  1. Resolves the target via the graph loader (callable, object, or string source)
  2. Captures the task in the observe phase
  3. Runs the target in the act phase
  4. Records the result in the reflect phase
  5. Emits all events to the timeline
import noesis as ns

# Your adapter is just a function
def my_adapter(task: str) -> str:
    return f"Processed: {task}"

# Noēsis wraps it with cognition
episode_id = ns.solve("Do something", using=my_adapter)

​
Resolution rules

using= accepts several shapes, resolved by the loader:
  • Callable: used directly (or invoked if it is a zero-arg factory).
  • Object: used directly if it exposes invoke() or is callable.
  • String: can be a dotted factory (pkg.mod:make), a filesystem path, or a short name resolved via flows.<name> or noesis_user.<name>.
Noēsis invokes adapters via invoke(), run(), or __call__() (in that order). Objects that only define execute() are not compatible with ns.solve() unless wrapped. If you need to map the input task string to a structured payload, you can:
  • wrap the target and do the mapping inside your wrapper, or
  • if supported by the invocation path you’re using, define a __noesis_input_mapper__ callable that transforms the task before invocation.

​
Plain functions

The simplest adapter is a plain Python function: 
import noesis as ns


def echo_adapter(task: str) -> dict:
    """Simple adapter that echoes the task."""
    return {"result": task.upper()}


episode_id = ns.solve("hello world", using=echo_adapter)
summary = ns.summary.read(episode_id)
print(summary["metrics"]["success"])  # 1 (no exception raised)

​
LangGraph integration

Compiled LangGraph graphs expose invoke(), so you can pass them directly: 
import noesis as ns
from langgraph.graph import StateGraph


# Define your LangGraph
def create_graph():
    graph = StateGraph(dict)
    
    def process(state):
        return {"result": state["task"].upper()}
    
    graph.add_node("process", process)
    graph.set_entry_point("process")
    graph.set_finish_point("process")
    
    return graph.compile()


# Create the graph
app = create_graph()


# Run through Noēsis
episode_id = ns.solve(
    "Process this request",
    using=app,
    intuition=True,
)
If your graph returns awaitables, wrap it with noesis.adapters.langgraph.LangGraphAdapter. It resolves awaitables with asyncio.run() when no event loop is running (and raises if a loop is already running). It also supports an optional input mapper and may fall back to __noesis_input_mapper__ if the wrapped graph provides one.

​
CrewAI integration

Wrap a CrewAI crew with noesis.adapters.crewai.CrewAIAdapter: 
import noesis as ns
from crewai import Agent, Task, Crew
from noesis.adapters.crewai import CrewAIAdapter


def create_crew():
    researcher = Agent(
        role="Researcher",
        goal="Research topics thoroughly",
        backstory="You are an expert researcher.",
    )
    
    task = Task(
        description="Research the given topic",
        agent=researcher,
    )
    
    return Crew(agents=[researcher], tasks=[task])


crew = create_crew()

episode_id = ns.solve(
    "Research quantum computing",
    using=CrewAIAdapter(crew),
    intuition=True,
)
Or wrap manually: 
def crewai_adapter(task: str) -> dict:
    # CrewAI expects an input payload; adapt as needed for your crew definition.
    result = crew.kickoff({"task": task})
    return {"result": str(result)}


episode_id = ns.solve(
    "Research quantum computing",
    using=crewai_adapter,
    intuition=True,
)
CrewAIAdapter defaults to mapping the task as {"task": task} unless the crew defines __noesis_input_mapper__. If your crew expects a different input shape, provide __noesis_input_mapper__ on the crew or pass input_mapper= when constructing CrewAIAdapter.

​
Claude Agent SDK integration

When you’re already running inside an event loop, use ns.solve_async(...) and pass an async adapter directly. 
import asyncio
import noesis as ns
from claude_agent_sdk import ClaudeAgentOptions, query


async def claude_adapter(task: str) -> object:
    last = None
    async for message in query(
        prompt=task,
        options=ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Bash"]),
    ):
        if hasattr(message, "result"):
            last = message.result
    return last


episode_id = await ns.solve_async(
    "Find and fix the bug in auth.py",
    using=claude_adapter,
    intuition=True,
)
If you’re in synchronous code, wrap the SDK’s async loop with asyncio.run(...) or move the call to a worker thread before calling ns.solve(...).
SDK surface shown above reflects the public claude_agent_sdk examples; verify against your installed SDK version.

​
Experimental Assistants adapter

noesis.adapters.assistant.AssistantsAdapter is experimental and writes its own events via write_event. It follows the execute(...) protocol but does not expose invoke() / run() / __call__(), so it is not compatible with ns.solve() without a wrapper. Use it only with a custom runner that explicitly calls execute().

​
Custom adapters with metadata

Adapters can return any object. Noēsis uses str(result) when constructing the action summary. If you return a dict, the summary will include its string representation. 
import noesis as ns
from dataclasses import dataclass
from typing import Any


@dataclass
class AdapterResult:
    status: str
    result: Any
    metadata: dict


class InstrumentedAdapter:
    """Adapter with timing and metadata."""
    
    def __init__(self, name: str):
        self.name = name
    
    def __call__(self, task: str) -> dict:
        import time
        
        start = time.time()
        
        # Your actual logic
        result = self._process(task)
        
        duration = time.time() - start
        
        return {"adapter": self.name, "duration_seconds": duration, "result": result}
    
    def _process(self, task: str) -> str:
        # Override in subclasses
        return task


# Usage
adapter = InstrumentedAdapter("my-adapter")
episode_id = ns.solve("task", using=adapter)

​
Error handling

Noēsis treats exceptions as failures. Raise to mark a failed action: 
import noesis as ns


def safe_adapter(task: str) -> dict:
    # Your logic
    result = process(task)
    if not result:
        raise ValueError("empty result")
    return {"result": result}


episode_id = ns.solve("risky task", using=safe_adapter)
summary = ns.summary.read(episode_id)

# Check status
if summary["metrics"]["success"] == 0:
    events = list(ns.events.read(episode_id))
    reflect_event = next(e for e in events if e["phase"] == "reflect")
    print(f"Failed: {reflect_event['payload'].get('reasons')}")  # ["adapter_error"]

​
Using string sources

You can also pass a string using= value that the loader resolves:
  • pkg.mod:factory to import and call a zero-arg factory
  • ./path/to/module.py to load a local module
  • name to load flows.name or noesis_user.name

​
Adapter best practices

Return structured data. Return JSON-serializable objects so summaries are readable.
Handle timeouts. Long-running adapters should implement timeouts to prevent hung episodes.
Keep adapters stateless. If you need state, use the Noēsis state artifact rather than adapter instance variables.

​
Testing adapters

Test adapters independently before integrating: 
import pytest


def test_adapter_success():
    result = echo_adapter("test input")
    assert result["result"] == "TEST INPUT"


def test_adapter_raises_on_error():
    def bad_adapter(task: str) -> object:
        raise ValueError("boom")

    with pytest.raises(ValueError):
        bad_adapter("x")

​
Next steps

Export metrics

Send adapter metrics to your observability stack.

Python API reference

Full API documentation for adapters.