Integrations¶
once.integrations.mcp ¶
Run MCP server tools through once.
Works with the official mcp SDK's FastMCP (1.x) and MCPServer
(2.x). It does not import mcp; it only reads the Context object the server
passes in.
from once.integrations.mcp import once_tool
@server.tool()
@once_tool(once, name="charge_card")
async def charge_card(amount: int, customer: str, ctx: Context) -> dict: ...
The Context parameter is left out of the key. The scope is, in order: the
value of scope_arg if you name one, the MCP session id (the
mcp-session-id header on streamable HTTP), or the enclosing
once.scope(). Request ids are not used, because a retried call is a new
request with a new id.
once_tool ¶
once_tool(once: Once, fn: Callable[P, R]) -> Callable[P, R]
once_tool(once: Once, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None, context_arg: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]
once_tool(once: Once, fn: Callable[..., Any] | None = None, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None, context_arg: str | None = None) -> Any
Decorate an MCP tool function so each distinct call runs at most once.
a tool parameter whose value is the scope (an order or run id
the model passes in). It stays part of the key too.
context_arg: the Context parameter, if it is not annotated as Context or
named ctx or context.
session_id ¶
session_id(ctx: Any) -> str | None
The MCP session id for a Context, or None outside a session.
once.integrations.langchain ¶
Run LangChain / LangGraph tools through once.
It does not import langchain. Put it under @tool and give the function a
config: RunnableConfig parameter; LangChain fills it in and hides it from
the model.
from once.integrations.langchain import once_tool
@tool
@once_tool(once)
def send_email(to: str, subject: str, config: RunnableConfig) -> str: ...
The config parameter is left out of the key. The scope is
config["configurable"]["thread_id"] (change it with scope_key), so a
LangGraph thread that replays a step after a crash or interrupt gets the
stored result. Without a thread id, the enclosing once.scope() applies.
once_tool ¶
once_tool(once: Once, fn: Callable[P, R]) -> Callable[P, R]
once_tool(once: Once, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_key: str = 'thread_id', config_arg: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]
once_tool(once: Once, fn: Callable[..., Any] | None = None, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_key: str = 'thread_id', config_arg: str | None = None) -> Any
Decorate a LangChain tool function so each distinct call runs at most once.
the RunnableConfig parameter, if it is not annotated as
RunnableConfig or named config.
config_scope ¶
config_scope(config: Any, scope_key: str = 'thread_id') -> str | None
Read configurable[scope_key] from a RunnableConfig (a dict).
once.integrations.openai_agents ¶
Run OpenAI Agents SDK function tools through once.
It does not import agents. Put it under @function_tool; the SDK passes
a RunContextWrapper or ToolContext as the first parameter and hides it
from the model.
from once.integrations.openai_agents import once_tool
@function_tool
@once_tool(once, name="charge_card")
async def charge_card(ctx: ToolContext[Deps], amount: int, customer: str) -> str: ...
The context parameter is left out of the key. The scope is, in order: the
value of scope_arg if you name one, scope_from(ctx) if given, the run's
RunConfig.group_id (set it to your conversation or session id), or the
enclosing once.scope(). Tool call ids are not used, because a retried call
gets a new one.
once_tool ¶
once_tool(once: Once, fn: Callable[P, R]) -> Callable[P, R]
once_tool(once: Once, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None, scope_from: ScopeFrom | None = None, context_arg: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]
once_tool(once: Once, fn: Callable[..., Any] | None = None, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None, scope_from: ScopeFrom | None = None, context_arg: str | None = None) -> Any
Decorate an Agents SDK tool function so each distinct call runs at most once.
a tool parameter whose value is the scope. It stays part of the
key too.
scope_from: called with the context wrapper, returns the scope or None,
e.g. lambda ctx: ctx.context.user_id.
context_arg: the context parameter, if it is not annotated as
RunContextWrapper or ToolContext or named ctx, context or
wrapper.
group_id ¶
group_id(ctx: Any) -> str | None
RunConfig.group_id for a ToolContext, or None if it is not set.
once.integrations.pydantic_ai ¶
Run Pydantic AI tools through once.
It does not import pydantic_ai. Put it under @agent.tool (or pass the
function in Agent(tools=[...])); Pydantic AI passes RunContext as the
first parameter and hides it from the model.
from once.integrations.pydantic_ai import once_tool
@agent.tool
@once_tool(once, scope_arg="order_id")
async def refund(ctx: RunContext[Deps], order_id: str, amount: int) -> str: ...
The RunContext parameter is left out of the key. The scope is, in order: the
value of scope_arg if you name one, scope_from(ctx) if given, the
enclosing once.scope(), then ctx.conversation_id (or ctx.run_id
with scope="run"). Pass your own conversation_id to agent.run:
Pydantic AI makes up a fresh one for every run that doesn't, and then each
run gets its own key. run_id is stable when a durable workflow replays a
run. Tool call ids and retry counts are not used.
once_tool ¶
once_tool(once: Once, fn: Callable[P, R]) -> Callable[P, R]
once_tool(once: Once, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None, scope_from: ScopeFrom | None = None, scope: RunScope = 'conversation', context_arg: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]
once_tool(once: Once, fn: Callable[..., Any] | None = None, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None, scope_from: ScopeFrom | None = None, scope: RunScope = 'conversation', context_arg: str | None = None) -> Any
Decorate a Pydantic AI tool function so each distinct call runs at most once.
a tool parameter whose value is the scope. It stays part of the
key too.
scope_from: called with the RunContext, returns the scope or None, e.g.
lambda ctx: ctx.deps.user_id.
scope: "conversation" (the default) or "run": the RunContext id used when
nothing above it gives a scope.
context_arg: the RunContext parameter, if it is not annotated as
RunContext or named ctx or context.
context_scope ¶
context_scope(ctx: Any, scope: RunScope) -> str | None
ctx.conversation_id or ctx.run_id from a RunContext.
once.integrations.crewai ¶
Run CrewAI tools through once.
It does not import crewai. Put it under @tool:
from once.integrations.crewai import once_tool
@tool("charge_card")
@once_tool(once, name="charge_card", scope_arg="order_id")
def charge_card(order_id: str, amount: int) -> str: ...
CrewAI wants a docstring on the function. Its tools get no run context, so
the scope is the value of scope_arg if you name one, otherwise the
enclosing once.scope(). Wrap crew.kickoff() in
with once.scope(run_id):; CrewAI copies context variables into its worker
threads, so the tools see it.
For a BaseTool subclass, decorate _run (or _arun) and pass a name.
self is left out of the key.
once_tool ¶
once_tool(once: Once, fn: Callable[P, R]) -> Callable[P, R]
once_tool(once: Once, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None) -> Callable[[Callable[P, R]], Callable[P, R]]
once_tool(once: Once, fn: Callable[..., Any] | None = None, /, *, name: str | None = None, timeout: float | None = None, ttl: TTL | None = None, verify: Verifier | None = None, scope_arg: str | None = None) -> Any
Decorate a CrewAI tool function so each distinct call runs at most once.
a tool parameter whose value is the scope (an order or run id
the model passes in). It stays part of the key too.