DeepAgents Adapter

`DeepAgentsAdapter`

Wraps a deepagents graph to apply stroma contract validation and cost tracking.

deepagents builds on LangGraph internally, so the compiled graph returned by deepagents.create_deep_agent() exposes the same node interface. This adapter discovers nodes decorated with @stroma_deepagents_node, replaces them with async validating wrappers, and returns the modified graph.

Checkpointing boundary: deepagents manages its own LangGraph checkpointer. Stroma does not inject checkpointing by default. Pass checkpoint_store only if you need a secondary stroma-managed checkpoint layer — you are then responsible for ensuring it does not conflict with deepagents' internal checkpointer.

Cost tracking: if a node returns a tuple matching stroma's 4-shape cost convention (dict, input_tokens, output_tokens, model), usage is recorded via the adapter's CostTracker. Plain dict returns record zero tokens.

Example

from stroma.contracts import ContractRegistry
from stroma.adapters.deepagents import DeepAgentsAdapter, stroma_deepagents_node

registry = ContractRegistry()
# ... register contracts, decorate nodes ...

adapter = DeepAgentsAdapter(registry)
agent = create_deep_agent(...)
wrapped = adapter.wrap(agent)
result = await wrapped.ainvoke({"text": "hello"})

`cost_tracker` `property`

The CostTracker accumulating usage across all wrapped node calls.

`wrap(agent)`

Discover stroma-decorated nodes in agent and replace them with validating wrappers.

agent is the compiled graph returned by deepagents.create_deep_agent(). Raises ImportError if deepagents is not installed.

`stroma_deepagents_node(node_id, contract)`

Decorator that attaches stroma contract metadata to a deepagents node function.

Binds node_id and contract as attributes on the decorated function so DeepAgentsAdapter can discover and validate it. Works identically to stroma_langgraph_node — deepagents nodes are LangGraph nodes under the hood.

Example

from pydantic import BaseModel
from stroma.contracts import NodeContract
from stroma.adapters.deepagents import stroma_deepagents_node

class Query(BaseModel):
    text: str

class Answer(BaseModel):
    response: str

contract = NodeContract(node_id="answer", input_schema=Query, output_schema=Answer)

@stroma_deepagents_node("answer", contract)
async def answer(state: Query) -> dict:
    return {"response": f"Got: {state.text}"}

DeepAgents Adapter