Runtime utilities

The tenxgraph.utils package holds the small helpers that 10xGraph applications use around the graph: graceful shutdown on SIGINT and SIGTERM, secret redaction for logs, calling sync or async functions, converting state to LLM message dicts, reading @tool metadata, and in-process metrics with optional OpenTelemetry export.

Graceful shutdown

Signal handling and controlled shutdown for asyncio applications. Use these helpers to finish in-flight work and clean up resources when the process receives SIGINT or SIGTERM.

GracefulShutdownManager

Manager for coordinating graceful shutdown of asyncio applications. It registers SIGINT and SIGTERM handlers on the event loop that set shutdown_requested and run your callbacks, and it protects critical sections from interruption. On Windows, loop signal handlers are not supported, so registration logs a warning instead of failing.

Example:

import asyncio
from tenxgraph.utils import GracefulShutdownManager


async def main():
    manager = GracefulShutdownManager(shutdown_timeout=10.0)
    manager.add_shutdown_callback(lambda: print("Signal received"))
    manager.register_signal_handlers()

    try:
        # Protect initialization from interruption
        with manager.protect_section():
            await asyncio.sleep(0.1)  # stand-in for real setup

        # Run until SIGINT or SIGTERM sets shutdown_requested
        await manager.wait_for_shutdown()
        print("Shutting down")
    finally:
        manager.unregister_signal_handlers()
        # Protect cleanup from a second signal
        with manager.protect_section():
            await asyncio.sleep(0.1)  # stand-in for real cleanup


asyncio.run(main())

DelayedKeyboardInterrupt

Context manager that delays SIGINT and SIGTERM handling. Use it to protect critical sections: the signal is recorded, and the original handler runs when the context exits. It installs handlers with signal.signal, so use it from the main thread only, and keep protected sections short.

Example:

from tenxgraph.utils import DelayedKeyboardInterrupt

with DelayedKeyboardInterrupt():
    # Critical section, for example writing a file in two steps
    print("writing")
# If SIGINT/SIGTERM was received, it is handled here

delayed_keyboard_interrupt()

Functional form of DelayedKeyboardInterrupt, built with contextlib.contextmanager. It takes no arguments and yields the DelayedKeyboardInterrupt instance.

Example:

from tenxgraph.utils import delayed_keyboard_interrupt

with delayed_keyboard_interrupt():
    # Critical section
    print("writing")

setup_exception_handler(loop)

Install an exception handler on the event loop that suppresses benign errors raised during shutdown (ConnectionResetError, and OSError with the Windows invalid-handle code). Any other exception is logged as an error with its traceback.

Example:

import asyncio
from tenxgraph.utils import setup_exception_handler

async def main():
    setup_exception_handler(asyncio.get_running_loop())
    await asyncio.sleep(0.1)


asyncio.run(main())

shutdown_with_timeout(coro_or_task, timeout, task_name)

Await a coroutine, task, or future for at most timeout seconds. A coroutine is wrapped in a task. On timeout the task is cancelled. The function never raises for these outcomes: it reports them in the returned dict.

Returns: dict[str, Any] with status ("completed", "timeout" or "error") and duration in seconds. The "error" result also has an error string.

Example:

import asyncio
from tenxgraph.utils import shutdown_with_timeout

async def cleanup():
    await asyncio.sleep(0.5)


async def main():
    result = await shutdown_with_timeout(cleanup(), timeout=10.0, task_name="cleanup")
    print(result)  # for example {'status': 'completed', 'duration': 0.5}


asyncio.run(main())

Logging and secret redaction

Redact credentials from log output. Redaction is heuristic, so treat it as a safety net and avoid logging secrets in the first place.

mask_secrets(text)

Redact common credential formats from a string. Masks OpenAI/Google/GitHub/Slack/AWS keys, Bearer tokens, key=value secrets, and signed-URL credential query parameters.

Returns: str with secrets replaced by ***REDACTED***. Empty input is returned unchanged. The Bearer scheme and key=value names are kept; only the value is replaced.

Example:

from tenxgraph.utils import mask_secrets

text = "Authorization: Bearer sk-proj-abc123xyz"
masked = mask_secrets(text)
print(masked)  # "Authorization: Bearer ***REDACTED***"

SecretRedactionFilter

logging.Filter subclass that redacts secrets from a record’s formatted message and always lets the record through. Add it to a handler to cover every logger that propagates to that handler. Adding it to a logger only covers records emitted directly on that logger, not its children.

Example:

import logging
from tenxgraph.utils import SecretRedactionFilter

handler = logging.StreamHandler()
handler.addFilter(SecretRedactionFilter())
logger = logging.getLogger("myapp")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

logger.info("Token: sk-1234567890abcdef")  # Token: ***REDACTED***

install_secret_redaction(logger_name)

Attach one SecretRedactionFilter to a logger and to each handler that logger has at call time. Call it after configuring handlers. For child loggers, add the filter to your handlers directly.

Returns: SecretRedactionFilter instance.

Example:

import logging
from tenxgraph.utils import install_secret_redaction

# Configure your logging first
logging.basicConfig(level=logging.INFO)

# Install redaction on the root logger and its handlers
install_secret_redaction("")
logging.getLogger("myapp").info("password=hunter2hunter2")  # password=***REDACTED***

Async and sync calling

Helpers for crossing the sync and async boundary. call_sync_or_async is for code already inside an event loop, and run_coroutine is for plain synchronous code.

call_sync_or_async(func, *args, **kwargs)

Call a function that may be sync or async and return its result. Async functions are awaited directly. Sync functions run in a worker thread through asyncio.to_thread, so they do not block the event loop, and an awaitable result is awaited before it is returned. This is an async def, so you must await it.

Returns: The result of the function call.

Example:

import asyncio
from tenxgraph.utils import call_sync_or_async

def sync_work(x: int) -> int:
    return x * 2

async def async_work(x: int) -> int:
    await asyncio.sleep(0.1)
    return x * 3

async def main():
    a = await call_sync_or_async(sync_work, 5)
    b = await call_sync_or_async(async_work, 5)
    print(a, b)  # 10 15

asyncio.run(main())

run_coroutine(func)

Run a coroutine from synchronous code and return its result. With no running loop it uses asyncio.run(). If a loop is already running in this thread it uses asyncio.run_coroutine_threadsafe() and blocks on the result, which deadlocks when called from the loop’s own thread, so call it from a different thread or a plain script.

Returns: The result of the coroutine.

Example:

import asyncio
from tenxgraph.utils import run_coroutine

async def async_task():
    await asyncio.sleep(0.1)
    return "done"

result = run_coroutine(async_task())
print(result)  # "done"

Message conversion

Build the list of message dicts you send to an LLM from system prompts, the state’s context, and any extra messages.

convert_messages(system_prompts, state, extra_messages)

Convert system prompts, agent state, and extra messages into one list of dicts. The order is: system prompts, then state.context_summary (as an assistant message) if set, then each message in state.context, then extra_messages. System prompt strings may use {field} placeholders, filled from state.model_dump(). If a placeholder names a field the state lacks, the original prompt is used and a warning is logged. Messages with multimodal content get OpenAI-style content parts.

Returns: list[dict[str, Any]] of message dicts ready for an LLM API.

Raises: ValueError if system_prompts is None.

Example:

from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils import convert_messages

state = AgentState(context=[Message.text_message("Hello")])
messages = convert_messages(
    system_prompts=[{"role": "system", "content": "You are helpful."}],
    state=state,
)
print(len(messages))  # 2 (system prompt + the context message)

Tool metadata

Read back the metadata that the @tool decorator attaches to a function. Both helpers are importable from tenxgraph.utils, as is tool itself.

get_tool_metadata(func)

Return the tool metadata stored on a function as a dict. It works on any callable: an undecorated function yields None values and an empty tags set.

Returns: dict with keys name, description, tags, provider, capabilities, metadata and parameters. tags is a set (empty when unset). The other values are None when unset.

Example:

from tenxgraph.utils import tool, get_tool_metadata

@tool(name="calculator", tags=["math"])
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

metadata = get_tool_metadata(add)
print(metadata["name"])      # "calculator"
print(metadata["tags"])      # {'math'}
print(metadata["description"])  # Add two numbers.

has_tool_decorator(func)

Return whether a function carries @tool metadata. Use it to separate decorated tools from plain callables.

Returns: bool.

Example:

from tenxgraph.utils import tool, has_tool_decorator

@tool
def decorated():
    """A tool."""
    pass

def not_decorated():
    pass

print(has_tool_decorator(decorated))      # True
print(has_tool_decorator(not_decorated))  # False

Metrics instrumentation

An in-process metrics registry with no required dependencies. Counters and timers live in a thread-safe module-level registry and can be exported through OpenTelemetry when you call setup_otel_metrics. These functions live in tenxgraph.utils.metrics, not tenxgraph.utils.

counter(name)

Get or create the counter registered under name. Repeated calls with the same name return the same object. Increment it with .inc(amount=1, attributes=None); attributes are used only for OpenTelemetry export.

Returns: Counter instance.

Example:

from tenxgraph.utils.metrics import counter

requests = counter("requests_total")
requests.inc()
requests.inc(5, attributes={"endpoint": "/api"})

timer(name, attributes)

Return a context manager that times a block in milliseconds. Attributes become OpenTelemetry histogram dimensions, and an outcome attribute ("ok" or "error") is added automatically. Exceptions raised in the block are not suppressed.

Returns: Context manager.

Example:

from tenxgraph.utils.metrics import timer

with timer("db_query_ms", attributes={"table": "users"}):
    total = sum(range(1000))  # stand-in for a real query

setup_otel_metrics(meter)

Bridge the registry to OpenTelemetry. Call it once at startup to export counters and timers through your configured MeterProvider. Install the otel extra to get OpenTelemetry.

Returns: bool. True if the bridge was installed; False if OpenTelemetry is not installed.

Example:

from tenxgraph.utils.metrics import counter, timer, setup_otel_metrics

# Enable OTEL export (returns False if OpenTelemetry is not installed)
installed = setup_otel_metrics()

# All counter/timer calls now export
counter("events_total").inc()
with timer("latency_ms"):
    total = sum(range(1000))  # stand-in for real work

snapshot()

Return a thread-safe copy of every registered metric.

Returns: dict with counters (name to integer value) and timers (name to a dict with count, avg_ms and max_ms).

Example:

from tenxgraph.utils.metrics import counter, snapshot

counter("total").inc(10)
print(snapshot())
# {'counters': {'total': 10}, 'timers': {}}

Import paths

All utilities are exported from tenxgraph.utils:

from tenxgraph.utils import (
    # Shutdown
    GracefulShutdownManager,
    DelayedKeyboardInterrupt,
    delayed_keyboard_interrupt,
    setup_exception_handler,
    shutdown_with_timeout,
    # Logging
    SecretRedactionFilter,
    install_secret_redaction,
    mask_secrets,
    # Async/sync
    call_sync_or_async,
    run_coroutine,
    # Messages
    convert_messages,
    # Tool metadata
    get_tool_metadata,
    has_tool_decorator,
)
from tenxgraph.utils.metrics import (
    counter,
    timer,
    setup_otel_metrics,
    snapshot,
)