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
ns.solve() does not call execute() directly.
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,
)
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,
)
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,
)
{"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().
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 modulename 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
