Integrate adapters

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:

Build ToolInvocationInput from inbound protocol data.
Call prepare_tool_invocation(...) with your ports and persist the returned draft.
If status is pending_approval, collect approval externally and persist ToolApprovalDecision using the same run_id + draft_id.
Call execute_prepared_tool_invocation(...) with that run_id + draft_id.
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

Symptom	Likely cause	Fix
`PreparedToolInvocationNotFoundError`	approval service called execute with wrong `run_id`/`draft_id`	persist and pass durable identity unchanged across systems
`ApprovalDecisionRequiredError`	decision missing or not `approved`	write an approved `ToolApprovalDecision` before execute
`ApprovalDecisionBindingError`	decision is not bound to reviewed prepared intent	store `request_id`, `reviewed_fingerprint`, and `impact_hash` from the prepared artifact and verify before save
`tool.replayed` result with no dispatch	same idempotency key and fingerprint seen previously	treat as successful replay; do not retry with a new side effect
`summary.json` / `manifest.json` missing while run is paused	expected non-terminal lifecycle state	inspect 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:

Confirm one pending draft exists in tool_invocations/prepared/ for the target run.
Persist an approved ToolApprovalDecision bound to the same run_id + draft_id.
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:

Resolves the target via the graph loader (callable, object, or string source)
Captures the task in the observe phase
Runs the target in the act phase
Records the result in the reflect phase
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")

Get started

Tutorials

How-to guides

Explanation

Integrate adapters

How adapters work

The `ns.solve()` integration contract

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

Protocol-first tool invocation contract (internal)

Intent

Prepare -> execute workflow

Event and identity invariants

Common pitfalls

Runtime bridge guardrails (current)

Subprocess payload contract

Operational runbook: inspect and resume pending drafts

Runtime evidence checks (pause/continue)

Resolution rules

Plain functions

LangGraph integration

CrewAI integration

Claude Agent SDK integration

Experimental Assistants adapter

Custom adapters with metadata

Error handling

Using string sources

Adapter best practices

Testing adapters

Next steps

Export metrics

Python API reference

​How adapters work

​The ns.solve() integration contract

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

​Protocol-first tool invocation contract (internal)

​Intent

​Prepare -> execute workflow

​Event and identity invariants

​Common pitfalls

​Runtime bridge guardrails (current)

​Subprocess payload contract

​Operational runbook: inspect and resume pending drafts

​Runtime evidence checks (pause/continue)

​Resolution rules

​Plain functions

​LangGraph integration

​CrewAI integration

​Claude Agent SDK integration

​Experimental Assistants adapter

​Custom adapters with metadata

​Error handling

​Using string sources

​Adapter best practices

​Testing adapters

​Next steps

Export metrics

Python API reference

How adapters work

The `ns.solve()` integration contract

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

Protocol-first tool invocation contract (internal)

Intent

Prepare -> execute workflow

Event and identity invariants

Common pitfalls

Runtime bridge guardrails (current)

Subprocess payload contract

Operational runbook: inspect and resume pending drafts

Runtime evidence checks (pause/continue)

Resolution rules

Plain functions

LangGraph integration

CrewAI integration

Claude Agent SDK integration

Experimental Assistants adapter

Custom adapters with metadata

Error handling

Using string sources

Adapter best practices

Testing adapters

Next steps