Threadify

Use the optional harnest-threadify package to link sessions to Threadify and coordinate shared business workflows.

The optional integration is maintained in packages/harnest-threadify in the Harnest repository. Contract permission waits require threadify-sdk>=0.4.0,<0.5.

Configure the agent

For development from the Harnest checkout, install into the environment running that checkout:

uv pip install --python .venv/bin/python --no-deps ./packages/harnest-threadify
uv pip install --python .venv/bin/python "threadify-sdk>=0.4.0,<0.5"

Copy the following into your existing agent’s lifecycle/telemetry.py. Keep its session storage configuration. Change service_name to a stable application/environment identity and select your business tools and attribute names.

Complete lifecycle configuration

lifecycle/telemetry.py

"""Copy into an agent's lifecycle/ directory to opt into Threadify."""

from harnest import context, lifecycle
from harnest_threadify import BusinessFilter, Threadify

threadify = Threadify(
    service_name="support-agent",
    business_tools=("lookup_order", "issue_refund"),
    filter=BusinessFilter(attributes=("order.id", "ticket.category")),
)

@lifecycle.telemetry_exporter
def telemetry():
    """Export only session-linked business spans to Threadify."""
    return threadify.telemetry_exporter()

@lifecycle.resource
@context.provider("threadify")
def connection():
    """Own the async SDK connection and make native access explicit."""
    return threadify

@lifecycle.session.created
async def session_created(info, session):
    """Create or recover the thread after session persistence succeeds."""
    await threadify.session_created(info, session)

@lifecycle.agent.before
async def invocation(active, request):
    """Link existing sessions and all supported invocation transports."""
    return await threadify.before_invocation(active, request)

@lifecycle.http.scope
def request_scope(active, request):
    """Keep request correlation active until streaming finishes."""
    return threadify.request_scope(active, request)

@lifecycle.http.after
def response(active, head):
    """Record handled HTTP failures without response data."""
    return threadify.after_http(active, head)

@lifecycle.tool.after
def tool_completed(active, result):
    """Export selected business-tool outcomes without result payloads."""
    return threadify.tool_completed(active, result)

@lifecycle.tool.on_error
def tool_failed(active, error):
    """Export selected business-tool failures without exception text."""
    threadify.tool_failed(active, error)

Provide THREADIFY_API_KEY through the server environment or deployment secret store, then start the agent. Compilation never reads the key or opens a connection.

Business data only

Exported	Excluded by default
Session-linked request outcome and timing	Health checks and requests without an agent session
Decision name, version, action, outcome, and duration	Decision input, questions, scores, and raw results
Outcomes of tools listed in `business_tools`	Skill discovery, model calls, and unselected tools
Explicit `business.*` spans and allowed scalar attributes	Prompts, replies, tool arguments/results, nested payloads
Thread ID, `sessionId`, and a scoped session reference	Raw user IDs, URLs, headers, exception messages, events, links, and framework resource metadata

ADK and LangGraph use the same filter. Exact attribute names are required; attribute wildcards are rejected. Strings are limited to 256 characters by default. Framework payload namespaces such as gen_ai.*, llm.*, and exception.* stay excluded even if named in the attribute allowlist. Other telemetry destinations receive their original spans. Use static business-step names and deliberately select non-sensitive attribute values:

from harnest import context
from harnest_threadify import Threadify

integration = context.resource("threadify", Threadify)
with integration.step("refund_approved", attributes={"order.id": "order-42"}):
    pass  # Perform the authored business action here.

Session ownership

Session links use a hash of application, user, and session identity. sessionId is a Threadify reference; equal IDs from different users remain separate. HTTP scopes cover streaming and cancellation; agent hooks cover other supported invocations. Native sessions are linked on first invocation. Later turns reuse the same Threadify thread. Route a session to one worker during thread creation. Creation is serialized within one integration instance, and ambiguous results require lookup before retry. Shared contract workflows instead use Threadify’s atomic application-owned key. Each instance tracks up to 10,000 linked or unresolved sessions. At capacity, new links are skipped. Missing credentials fail startup; remote failures let the agent continue. A failed startup connection requires restart. Export is best effort, not a durable audit ledger.

Native SDK access

integration = context.resource("threadify", Threadify)
connection = integration.native             # Native SDK Connection
NativeThreadify = integration.native_class  # Native SDK factory class
thread = await integration.current_thread() # Native session ThreadInstance, or None

Use native async operations on the application’s event loop. Native SDK operations are explicit application actions and bypass the telemetry filter. Completing a native thread ends that business workflow; the integration does not automatically start a replacement for the same session.

Check a shared workflow before a tool call

The workflow owner opens a versioned contract with workflow(key, contract=..., role=...) and returns the thread ID after authorizing the user. The agent joins with join_workflow(thread_id, role=...). Both require a working Threadify connection. For an action with a side effect, use wait_for before calling the backend. Threadify’s can query previews eligibility; it does not claim an invocation. Pass a persisted invocation ID to WaitOptions and use that ID as the backend’s idempotency key.

from threadify import WaitOptions
from harnest import context
from harnest_threadify import Threadify

async def issue_refund(order, invocation_id, payments):
    integration = context.resource("threadify", Threadify)
    workflow = await integration.join_workflow(order.thread_id, role="processor")
    grant = await workflow.wait_for(
        "refund_issued", WaitOptions(timeout=10, invocation_id=invocation_id)
    )
    receipt = await payments.refund(order, idempotency_key=grant.invocation_id)
    await workflow.step("refund_issued").idempotency_key(
        grant.invocation_id
    ).add_context({"receipt": receipt.reference}).success(wait_for=True)
    return receipt

Report the backend’s actual outcome after the claimed action. Reconcile ambiguous responses by idempotency key before retrying. Recover a timed-out report validation with wait_for_validation and its step ID. Try examples/threadify-refund for an order-service prerequisite, human approval, a claimed refund invocation, and an idempotent backend. It uses synthetic in-memory orders; replace the demo backend with authorized services for real refunds. To try the project from the Harnest checkout, create its contract in Threadify and supply API keys allowed to report as orders and processor. The development Harnest principal is _harnest_neutral; use that value when preparing the synthetic order unless you configure authentication.

threadify-cli contracts preview --file examples/threadify-refund/contracts/refund.feature
threadify-cli contracts create --file examples/threadify-refund/contracts/refund.feature

export THREADIFY_ENGINE_URL=http://127.0.0.1:8081
export REFUND_API_URL=http://127.0.0.1:8091
export REFUND_API_TOKEN=local-demo-key
export THREADIFY_ORDERS_API_KEY=your-orders-key
export THREADIFY_API_KEY=your-processor-key

.venv/bin/python -m uvicorn backend:app --app-dir examples/threadify-refund --host 127.0.0.1 --port 8091
# In another terminal with the same environment:
.venv/bin/python examples/threadify-refund/prepare.py O-42 _harnest_neutral
harnest env sync examples/threadify-refund
harnest serve examples/threadify-refund

Set OPENAI_MODEL, OPENAI_BASE_URL, and any required model key before serving. The example reads these values from the environment. Complete the Harnest approval request when the agent calls issue_refund.

OTLP/HTTP Protobuf

The default transport uses Threadify’s native SDK. If your deployment supplies a standard OTLP/HTTP traces endpoint, select it when constructing the class:

threadify = Threadify(
    service_name="support-agent",
    otlp_endpoint=os.environ["THREADIFY_OTLP_TRACES_ENDPOINT"],
    otlp_headers={"Authorization": f"Bearer {os.environ['THREADIFY_OTLP_TOKEN']}"},
    filter=BusinessFilter(attributes=("order.id",)),
)

Import os and configure your receiver’s exact URL and headers. Filtering runs before serialization; the receiver must honor threadify.thread_id and session references. The SDK still manages session links and native access.

API reference

Reference: harnest_threadify.

Overview

Get started

Harnest Studio

Guides

API reference

Configure the agent

Business data only

Session ownership

Native SDK access

Check a shared workflow before a tool call

OTLP/HTTP Protobuf

API reference

​Configure the agent

​Business data only

​Session ownership

​Native SDK access

​Check a shared workflow before a tool call

​OTLP/HTTP Protobuf

​API reference

Configure the agent

Business data only

Session ownership

Native SDK access

Check a shared workflow before a tool call

OTLP/HTTP Protobuf

API reference