LangSmith Spans Reference

Helpers for emitting Nevermined-flavored payment spans into a LangSmith trace.

When the optional [langsmith] extra is installed and a LangSmith run is active in the current context (typically because LANGSMITH_TRACING=true is set and the call is inside a traced runnable), the @requires_payment decorator automatically uses these helpers to emit dedicated nvm:verify and nvm:settlement child spans nested under the active tool span.

For non-LangChain code paths (e.g. the FastAPI middleware) the context managers can also be used directly — see the example in verify_span / settlement_span.

All helpers in this module silently no-op when:

• the langsmith package is not installed, or
• no LangSmith run tree is active in the current context.

Failures inside this module never propagate out — observability is best-effort and must not interfere with the payment flow.

spans

LangSmith span helpers for Nevermined payment events.

All helpers in this module silently no-op when

• the langsmith package is not installed; or
• no LangSmith run tree is active in the current context (e.g. LANGSMITH_TRACING is unset, or the call is not inside a traced runnable).

Failures inside this module never propagate out -- observability is best-effort and must not interfere with the payment flow.

verify_span

verify_span(
    plan_ids: list[str],
    scheme: Optional[str] = None,
    network: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> Iterator[Optional[Any]]

Open an nvm:verify child span around a verify call.

Yields the underlying RunTree if LangSmith is active and a parent run is in scope; yields None otherwise. Always safe to call.

Exceptions raised by the caller's body propagate out unchanged -- the underlying langsmith.trace context manager sees them via its __exit__ and records the span as failed before re-raising. Only errors raised inside the trace SETUP (constructing or entering the LangSmith context) are swallowed, since those are pure observability concerns and must never interfere with the payment flow.

Example::

with verify_span(plan_ids=["plan-1"]) as span:
    verification = payments.facilitator.verify_permissions(...)
    add_metadata(span, build_verify_metadata(
        plan_ids=["plan-1"], verification=verification,
    ))

Source code in payments_py/langsmith/spans.py

@contextmanager
def verify_span(
    plan_ids: list[str],
    scheme: Optional[str] = None,
    network: Optional[str] = None,
    agent_id: Optional[str] = None,
) -> Iterator[Optional[Any]]:
    """Open an ``nvm:verify`` child span around a verify call.

    Yields the underlying ``RunTree`` if LangSmith is active and a parent
    run is in scope; yields ``None`` otherwise. Always safe to call.

    Exceptions raised by the caller's body propagate out unchanged -- the
    underlying ``langsmith.trace`` context manager sees them via its
    ``__exit__`` and records the span as failed before re-raising. Only
    errors raised inside the trace SETUP (constructing or entering the
    LangSmith context) are swallowed, since those are pure observability
    concerns and must never interfere with the payment flow.

    Example::

        with verify_span(plan_ids=["plan-1"]) as span:
            verification = payments.facilitator.verify_permissions(...)
            add_metadata(span, build_verify_metadata(
                plan_ids=["plan-1"], verification=verification,
            ))
    """
    inputs: dict = {"plan_ids": list(plan_ids)}
    if scheme:
        inputs["scheme"] = scheme
    if network:
        inputs["network"] = network
    if agent_id:
        inputs["agent_id"] = agent_id

    with _open_nvm_span("nvm:verify", inputs) as span:
        yield span

settlement_span

settlement_span(
    plan_ids: list[str], agent_id: Optional[str] = None
) -> Iterator[Optional[Any]]

Open an nvm:settlement child span around a settle call.

Same semantics as :func:verify_span.

Source code in payments_py/langsmith/spans.py

@contextmanager
def settlement_span(
    plan_ids: list[str],
    agent_id: Optional[str] = None,
) -> Iterator[Optional[Any]]:
    """Open an ``nvm:settlement`` child span around a settle call.

    Same semantics as :func:`verify_span`.
    """
    inputs: dict = {"plan_ids": list(plan_ids)}
    if agent_id:
        inputs["agent_id"] = agent_id

    with _open_nvm_span("nvm:settlement", inputs) as span:
        yield span

build_verify_metadata

build_verify_metadata(
    plan_ids: list[str],
    scheme: Optional[str] = None,
    network: Optional[str] = None,
    agent_id: Optional[str] = None,
    verification: Optional[VerifyResponse] = None,
    duration_ms: Optional[float] = None,
    token: Optional[str] = None,
) -> dict

Build the nvm.* metadata dict for a verify span. Drops None values.

token is abbreviated via :func:abbreviate_token before being surfaced as nvm.payment_token so the full credential never ends up in metadata that we control.

Source code in payments_py/langsmith/spans.py

def build_verify_metadata(
    plan_ids: list[str],
    scheme: Optional[str] = None,
    network: Optional[str] = None,
    agent_id: Optional[str] = None,
    verification: Optional[VerifyResponse] = None,
    duration_ms: Optional[float] = None,
    token: Optional[str] = None,
) -> dict:
    """Build the ``nvm.*`` metadata dict for a verify span. Drops ``None`` values.

    ``token`` is abbreviated via :func:`abbreviate_token` before being
    surfaced as ``nvm.payment_token`` so the full credential never ends
    up in metadata that we control.
    """
    md: dict = {"nvm.plan_ids": list(plan_ids)}
    if scheme:
        md["nvm.scheme"] = scheme
    if network:
        md["nvm.network"] = network
    if agent_id:
        md["nvm.agent_id"] = agent_id
    if duration_ms is not None:
        md["nvm.verify.duration_ms"] = round(duration_ms, 2)
    abbreviated = abbreviate_token(token)
    if abbreviated:
        md["nvm.payment_token"] = abbreviated
    if verification is not None:
        if verification.payer:
            md["nvm.payer"] = verification.payer
        if verification.network and "nvm.network" not in md:
            md["nvm.network"] = verification.network
        if verification.agent_request_id:
            md["nvm.agent_request_id"] = verification.agent_request_id
    return md

build_settle_metadata

build_settle_metadata(
    settlement: SettleResponse,
    plan_ids: list[str],
    agent_id: Optional[str] = None,
    duration_ms: Optional[float] = None,
    token: Optional[str] = None,
) -> dict

Build the nvm.* metadata dict for a settlement span. Drops None values.

token is abbreviated via :func:abbreviate_token before being surfaced as nvm.payment_token.

Source code in payments_py/langsmith/spans.py

def build_settle_metadata(
    settlement: SettleResponse,
    plan_ids: list[str],
    agent_id: Optional[str] = None,
    duration_ms: Optional[float] = None,
    token: Optional[str] = None,
) -> dict:
    """Build the ``nvm.*`` metadata dict for a settlement span. Drops ``None`` values.

    ``token`` is abbreviated via :func:`abbreviate_token` before being
    surfaced as ``nvm.payment_token``.
    """
    md: dict = {"nvm.plan_ids": list(plan_ids)}
    if agent_id:
        md["nvm.agent_id"] = agent_id
    if duration_ms is not None:
        md["nvm.settle.duration_ms"] = round(duration_ms, 2)
    abbreviated = abbreviate_token(token)
    if abbreviated:
        md["nvm.payment_token"] = abbreviated
    if settlement.credits_redeemed is not None:
        md["nvm.credits_redeemed"] = settlement.credits_redeemed
    if settlement.remaining_balance is not None:
        md["nvm.balance.after"] = settlement.remaining_balance
    if settlement.transaction:
        md["nvm.tx_hash"] = settlement.transaction
    if settlement.network:
        md["nvm.network"] = settlement.network
    if settlement.payer:
        md["nvm.payer"] = settlement.payer
    return md

active_run_tree

active_run_tree() -> Optional[Any]

Return the current LangSmith RunTree, or None if no run is active.

Safe to call regardless of whether langsmith is installed.

Source code in payments_py/langsmith/spans.py

def active_run_tree() -> Optional[Any]:
    """Return the current LangSmith ``RunTree``, or ``None`` if no run is active.

    Safe to call regardless of whether ``langsmith`` is installed.
    """
    if not _LANGSMITH_AVAILABLE:
        return None
    try:
        return _get_current_run_tree()
    except Exception:
        return None

add_metadata

add_metadata(
    run_tree: Optional[Any], metadata: dict
) -> None

Attach metadata to run_tree, swallowing any error.

No-op if run_tree is None or metadata is empty.

Source code in payments_py/langsmith/spans.py

def add_metadata(run_tree: Optional[Any], metadata: dict) -> None:
    """Attach ``metadata`` to ``run_tree``, swallowing any error.

    No-op if ``run_tree`` is ``None`` or ``metadata`` is empty.
    """
    if run_tree is None or not metadata:
        return
    try:
        run_tree.add_metadata(metadata)
    except Exception:
        logger.debug("LangSmith add_metadata failed (ignored)", exc_info=True)