Framework helpers

Five helpers for attributing work inside an agent framework. All of them do one thing: open an attribution scope for the duration of a node, task, or chain. The provider patches still capture the call itself, so nothing is counted twice.

None of them import the framework they are named after, so they are safe to use in a codebase where that framework is absent. For worked examples per framework, see Frameworks.

framework_unit()

capsera.framework_unit(
    *,
    agent: str,
    team: str | None = None,
    task_type: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
)

Returns a decorator for any framework node, step, or task. Keyword-only. Handles sync and async functions, and pops its scope in a finally.

The three helpers below call this one. Use it directly for a framework that has no named helper, or when you want the parameter names to read as generic.

langgraph_node()

capsera.langgraph_node(
    name: str,
    *,
    team: str | None = None,
    task_type: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
)

name is positional; everything else is keyword-only.

from capsera import langgraph_node

@langgraph_node("planner", team="research")
def plan(state): ...

crewai_agent()

capsera.crewai_agent(
    name: str,
    *,
    team: str | None = None,
    task_type: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
)

The same shape, for a CrewAI agent entrypoint.

crewai_task()

capsera.crewai_task(
    name: str,
    *,
    team: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
)

Note the missing parameter: crewai_task() sets task_type="crewai_task" itself and takes no task_type argument. If you need a task type of your own, use framework_unit().

LangChainCapseraCallback

capsera.LangChainCapseraCallback(
    *,
    agent: str,
    team: str | None = None,
    task_type: str | None = "langchain",
    session_id: str | None = None,
    customer_id: str | None = None,
    cost_center: str | None = None,
)

A callback handler for LangChain's callback protocol. Keyword-only, and the only helper here that takes session_id.

callback = capsera.LangChainCapseraCallback(agent="rag_chain", team="search")
chain.invoke({"question": "..."}, config={"callbacks": [callback]})

It pushes its scope on on_llm_start and pops it on on_llm_end and on_llm_error, tracking depth so an unbalanced callback sequence cannot pop past the bottom of the stack. The three handler methods accept arbitrary arguments, which is deliberate: LangChain's callback signatures have changed across versions, and a signature mismatch would otherwise raise inside your chain.

Reach for the decorators when the unit of work is a function you own, and for this when the work is a chain assembled at runtime.

What these helpers do not record

They set the same attribution fields as agent() and nothing more. There is no framework name on the event — a call made through LangGraph is distinguishable by the agent you named and by caller_file, not by a dedicated field.