> ## Documentation Index
> Fetch the complete documentation index at: https://noesis-32c1d602-cursor-technical-documentation-improvements.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
# Integrate adapters
> How to connect Noēsis to LangGraph, CrewAI, or your own agent runtime.
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:
```python theme={null}
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,
)
```
### Episode runtime bridge
Use the runtime bridge when a prepared tool invocation should participate in the
same episode lifecycle as `ns.run(...)`, `ns.solve(...)`, and `ns.resume_run(...)`.
The bridge is internal and intentionally narrow: it currently supports
`ToolProtocol.SUBPROCESS` only.
```python theme={null}
from noesis.usecases.tool_invocation.runtime_bridge import (
ToolRuntimeBridgePorts,
build_tool_invocation_actuation_bindings,
)
ports = ToolRuntimeBridgePorts(
prepared_repository=prepared_repository,
approval_repository=approval_repository,
idempotency_store=idempotency_store,
dispatch=subprocess_dispatch,
normalizer=normalizer,
authenticator=authenticator,
authorizer=authorizer,
preflight=preflight,
)
bindings = build_tool_invocation_actuation_bindings(
request_factory=lambda run_id: ToolInvocationInput(
run_id=run_id,
request_id="req-apply-canary",
draft_id=f"draft:{run_id}:req-apply-canary",
protocol=ToolProtocol.SUBPROCESS,
tool=ToolIdentity(namespace="deploy", name="apply_canary", version="1"),
raw_payload={"argv": ["./scripts/apply-canary.sh"]},
execution=ExecutionContext(timeout_ms=30_000, retry_limit=0),
security=SecurityContext(
principal_id="user:ops",
scopes=("deploy:write",),
policy_scope="prod/canary",
),
governance=GovernanceContext(
effect_kind=EffectKind.WRITE,
risk_tier=RiskTier.HIGH,
requires_approval=True,
),
),
run_dir=run_dir,
ports=ports,
run_lifecycle=run_lifecycle,
)
```
The prepare side of the bridge:
* creates canonical `action_candidate` evidence before any side effect
* persists a pending prepared draft for `run_id + draft_id`
* returns an interrupted actuation result when approval is required
* lets the run lifecycle emit `run.interrupt`, `run.checkpoint`, and
`run.state_projection`
After an approval service writes a bound `ToolApprovalDecision`, call
`ns.resume_run(episode_id, checkpoint_id=...)`. `resume_run(...)` detects a
pending prepared draft for the same run, builds resumed bindings internally, and
executes that draft without calling the original `request_factory` again.
### 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`
When the runtime bridge pauses a run, the episode trace also contains:
`action_candidate -> action.candidate_emitted -> tool.approval.pending -> run.interrupt -> run.checkpoint -> run.state_projection`
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`
For `ns.resume_run(...)` continuation, the existing event log remains an
append-only prefix and resume adds:
`run.resume -> tool.execution.started -> ... -> terminate`
Identity and binding rules:
* execute lookup key is `run_id + draft_id`
* runtime resume lookup expects exactly one pending draft for the run
* missing prepared draft raises `PreparedToolInvocationNotFoundError`
* multiple pending drafts raise `AmbiguousPreparedToolInvocationError`
* missing/non-approved decision raises `ApprovalDecisionRequiredError`
* mismatched `request_id`, reviewed fingerprint, or impact hash raises `ApprovalDecisionBindingError`
* non-subprocess runtime bridge protocols raise `UnsupportedToolProtocolError`
* 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 |
| `UnsupportedToolProtocolError` before draft files exist | runtime bridge prepare received `http` or `mcp` | use `ToolProtocol.SUBPROCESS` for bridge continuation, or call lower-level prepare/execute outside the bridge |
| `UnsupportedToolProtocolError` during `resume_run` with no `run.resume` event | pending draft has an unsupported protocol | correct/remove the draft before resuming; unsupported drafts fail before lifecycle mutation |
| `AmbiguousPreparedToolInvocationError` | more than one `pending_approval` draft exists for the same run | resolve or archive extra drafts so one pending draft remains |
| `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
* unsupported protocols fail before prepared-draft persistence on prepare
* unsupported pending drafts fail before `run.resume` is emitted on resume
* 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
```python theme={null}
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.` or `noesis_user.`.
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:
```python theme={null}
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:
```python theme={null}
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`:
```python theme={null}
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:
```python theme={null}
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.
```python theme={null}
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.
```python theme={null}
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:
```python theme={null}
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:
```python theme={null}
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
Send adapter metrics to your observability stack.
Full API documentation for adapters.