harnest.lifecycle

All Python modules · Import and compatibility guidance

HTTP agent principal grants

@lifecycle.http.agent_principal registers one sync or async function that receives a ConnectionContext and an AuthPrincipal. Return an AgentRuntimePrincipal to grant permissions or None to deny the invocation. See Grant permissions on built-in routes.

Hook control flow

Return	Effect
`context.next()`	Continue unchanged
`context.next(replacement)`	Pass a replacement to the next stage
`context.finish(result)`	Stop the chain with a final result

Exports

Name	Kind
`CoverageLevel`	Class
`DROP_EVENT`	Namespace
`EvalCaptureOptions`	Class
`Finish`	Class
`LifecycleContext`	Class
`LifecycleCoverage`	Class
`LifecycleListener`	Class
`LifecycleRegistration`	Class
`ModelCallRequest`	Class
`ModelCallResponse`	Class
`ModelLifecycleContext`	Class
`ModelMessage`	Class
`Next`	Class
`TransitionContext`	Class
`adk_plugin`	Function
`agent`	Namespace
`authenticate`	Namespace
`coverage`	Function
`credential_provider`	Namespace
`eval_capture_options`	Function
`evals`	Namespace
`http`	Namespace
`http_routes`	Namespace
`langgraph_middleware`	Function
`lifecycle_coverage`	Function
`mcp`	Namespace
`model`	Namespace
`output_policy`	Namespace
`registrations_for`	Function
`resource`	Namespace
`session`	Namespace
`skills`	Namespace
`storage`	Namespace
`telemetry_exporter`	Namespace
`tool`	Namespace

CoverageLevel

Describe how completely Harnest governs one runtime boundary.

CoverageLevel(*values)

Member	Value
`FULL`	`'full'`
`WRAPPED_ONLY`	`'wrapped-only'`
`BEST_EFFORT`	`'best-effort'`
`FRAMEWORK_OWNED`	`'framework-owned'`
`UNAVAILABLE`	`'unavailable'`

Example

from harnest.lifecycle import CoverageLevel

value = CoverageLevel.FULL

DROP_EVENT

Example

from harnest import lifecycle

value = lifecycle.DROP_EVENT

EvalCaptureOptions

Authored capture settings; None defers to the environment, then defaults.

EvalCaptureOptions(store: str, sample_rate: float | None = None) -> None

Fields (— means no declared default):

Field	Type	Default
`store`	`str`	`—`
`sample_rate`	`float \| None`	`None`

Parameters and example

Parameter	Type	Default
`store`	`str`	Required
`sample_rate`	`float \| None`	`None`

from harnest.lifecycle import EvalCaptureOptions

options = EvalCaptureOptions(store=...)

Finish

Complete an intercepting lifecycle stage without calling later work.

Finish(result: ResultT) -> None

Fields (— means no declared default):

Field	Type	Default
`result`	`ResultT`	`—`

Parameters and example

Parameter	Type	Default
`result`	`ResultT`	Required

from harnest.lifecycle import Finish

finish = Finish(result=...)

LifecycleContext

Portable invocation context shared by listeners for one execution.

LifecycleContext(
    framework: str,
    agent_name: str,
    invocation_id: str,
    user_id: str,
    session_id: str,
    metadata: Mapping[str, Any] = ...,
    attributes: MutableMapping[str, Any] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`framework`	`str`	`—`
`agent_name`	`str`	`—`
`invocation_id`	`str`	`—`
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`metadata`	`Mapping[str, Any]`	`dict()`
`attributes`	`MutableMapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`framework`	`str`	Required
`agent_name`	`str`	Required
`invocation_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`metadata`	`Mapping[str, Any]`	`...`
`attributes`	`MutableMapping[str, Any]`	`...`

from harnest.lifecycle import LifecycleContext

context = LifecycleContext(
    framework=...,
    agent_name=...,
    invocation_id=...,
    user_id=...,
    session_id=...,
)

LifecycleContext.next

LifecycleContext.next(replacement: Any = UNCHANGED) -> Next[Any]

Continue normally or pass a replacement to the next interceptor.

Parameters and example

Parameter	Type	Default
`replacement`	`Any`	`UNCHANGED`

from harnest.lifecycle import LifecycleContext

context: LifecycleContext
result = context.next()

LifecycleContext.finish

LifecycleContext.finish(result: ResultT) -> Finish[ResultT]

Return a final value without advancing to later execution.

Parameters and example

Parameter	Type	Default
`result`	`ResultT`	Required

from harnest.lifecycle import LifecycleContext

context: LifecycleContext
result = context.finish(result=...)

LifecycleCoverage

Stable report of lifecycle boundaries Harnest can actually guarantee.

LifecycleCoverage(
    framework: Literal['adk', 'langgraph'],
    mode: Literal['managed', 'advanced'],
    application: CoverageLevel,
    http: CoverageLevel,
    invocation: CoverageLevel,
    credentials: CoverageLevel,
    session: CoverageLevel,
    assets: CoverageLevel,
    model: CoverageLevel,
    tool: CoverageLevel,
    mcp: CoverageLevel,
    agent: CoverageLevel,
    subagent: CoverageLevel,
    checkpoint: CoverageLevel,
) -> None

Fields (— means no declared default):

Field	Type	Default
`framework`	`Literal['adk', 'langgraph']`	`—`
`mode`	`Literal['managed', 'advanced']`	`—`
`application`	`CoverageLevel`	`—`
`http`	`CoverageLevel`	`—`
`invocation`	`CoverageLevel`	`—`
`credentials`	`CoverageLevel`	`—`
`session`	`CoverageLevel`	`—`
`assets`	`CoverageLevel`	`—`
`model`	`CoverageLevel`	`—`
`tool`	`CoverageLevel`	`—`
`mcp`	`CoverageLevel`	`—`
`agent`	`CoverageLevel`	`—`
`subagent`	`CoverageLevel`	`—`
`checkpoint`	`CoverageLevel`	`—`

Parameters and example

Parameter	Type	Default
`framework`	`Literal['adk', 'langgraph']`	Required
`mode`	`Literal['managed', 'advanced']`	Required
`application`	`CoverageLevel`	Required
`http`	`CoverageLevel`	Required
`invocation`	`CoverageLevel`	Required
`credentials`	`CoverageLevel`	Required
`session`	`CoverageLevel`	Required
`assets`	`CoverageLevel`	Required
`model`	`CoverageLevel`	Required
`tool`	`CoverageLevel`	Required
`mcp`	`CoverageLevel`	Required
`agent`	`CoverageLevel`	Required
`subagent`	`CoverageLevel`	Required
`checkpoint`	`CoverageLevel`	Required

from harnest.lifecycle import LifecycleCoverage

coverage = LifecycleCoverage(
    framework=...,
    mode=...,
    application=...,
    http=...,
    invocation=...,
    credentials=...,
    session=...,
    assets=...,
    model=...,
    tool=...,
    mcp=...,
    agent=...,
    subagent=...,
    checkpoint=...,
)

LifecycleCoverage.stage

LifecycleCoverage.stage(name: str) -> CoverageLevel

Return one declared stage without exposing implementation state.

Parameters and example

Parameter	Type	Default
`name`	`str`	Required

from harnest.lifecycle import LifecycleCoverage

coverage: LifecycleCoverage
level = coverage.stage(name=...)

LifecycleCoverage.report

LifecycleCoverage.report() -> Mapping[str, str]

Return an immutable JSON-compatible diagnostics report.

Example

from harnest.lifecycle import LifecycleCoverage

coverage: LifecycleCoverage
result = coverage.report()

LifecycleCoverage.with_overrides

LifecycleCoverage.with_overrides(overrides: Mapping[str, CoverageLevel]) -> 'LifecycleCoverage'

Record adapter-observed limits without changing mode defaults.

Parameters and example

Parameter	Type	Default
`overrides`	`Mapping[str, CoverageLevel]`	Required

from harnest.lifecycle import LifecycleCoverage

coverage: LifecycleCoverage
coverage = coverage.with_overrides(overrides=...)

LifecycleListener

One discovered callback with a deterministic source identity.

LifecycleListener(
    phase: str,
    callback: Callable[..., Any],
    order: int,
    relative_path: str,
    line: int,
    function_name: str,
    framework: str | None = None,
    context_name: str | None = None,
    registration_name: str | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`phase`	`str`	`—`
`callback`	`Callable[..., Any]`	`—`
`order`	`int`	`—`
`relative_path`	`str`	`—`
`line`	`int`	`—`
`function_name`	`str`	`—`
`framework`	`str \| None`	`None`
`context_name`	`str \| None`	`None`
`registration_name`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`phase`	`str`	Required
`callback`	`Callable[..., Any]`	Required
`order`	`int`	Required
`relative_path`	`str`	Required
`line`	`int`	Required
`function_name`	`str`	Required
`framework`	`str \| None`	`None`
`context_name`	`str \| None`	`None`
`registration_name`	`str \| None`	`None`

from harnest.lifecycle import LifecycleListener

listener = LifecycleListener(
    phase=...,
    callback=...,
    order=...,
    relative_path=...,
    line=...,
    function_name=...,
)

LifecycleListener.identity

Return the stable identity used to order and deduplicate listeners.

Example

from harnest.lifecycle import LifecycleListener

listener: LifecycleListener
value = listener.identity

LifecycleRegistration

Metadata attached by one lifecycle decorator.

LifecycleRegistration(phase: str, order: int, framework: str | None = None, name: str | None = None) -> None

Fields (— means no declared default):

Field	Type	Default
`phase`	`str`	`—`
`order`	`int`	`—`
`framework`	`str \| None`	`None`
`name`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`phase`	`str`	Required
`order`	`int`	Required
`framework`	`str \| None`	`None`
`name`	`str \| None`	`None`

from harnest.lifecycle import LifecycleRegistration

registration = LifecycleRegistration(phase=..., order=...)

ModelCallRequest

Portable model input that a listener may replace or short-circuit.

ModelCallRequest(model: str | None, messages: tuple[ModelMessage, ...]) -> None

Fields (— means no declared default):

Field	Type	Default
`model`	`str \| None`	`—`
`messages`	`tuple[ModelMessage, ...]`	`—`

Parameters and example

Parameter	Type	Default
`model`	`str \| None`	Required
`messages`	`tuple[ModelMessage, ...]`	Required

from harnest.lifecycle import ModelCallRequest

request = ModelCallRequest(model=..., messages=...)

ModelCallResponse

Portable visible model output.

ModelCallResponse(text: str) -> None

Fields (— means no declared default):

Field	Type	Default
`text`	`str`	`—`

Parameters and example

Parameter	Type	Default
`text`	`str`	Required

from harnest.lifecycle import ModelCallResponse

response = ModelCallResponse(text=...)

ModelLifecycleContext

Stable provider-neutral facts associated with one model call.

ModelLifecycleContext(
    framework: str,
    agent_name: str | None = None,
    invocation_id: str | None = None,
    user_id: str | None = None,
    session_id: str | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`framework`	`str`	`—`
`agent_name`	`str \| None`	`None`
`invocation_id`	`str \| None`	`None`
`user_id`	`str \| None`	`None`
`session_id`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`framework`	`str`	Required
`agent_name`	`str \| None`	`None`
`invocation_id`	`str \| None`	`None`
`user_id`	`str \| None`	`None`
`session_id`	`str \| None`	`None`

from harnest.lifecycle import ModelLifecycleContext

context = ModelLifecycleContext(framework=...)

ModelLifecycleContext.next

ModelLifecycleContext.next(replacement: Any = UNCHANGED) -> Next[Any]

Continue normally or pass a replacement to the next interceptor.

Parameters and example

Parameter	Type	Default
`replacement`	`Any`	`UNCHANGED`

from harnest.lifecycle import ModelLifecycleContext

context: ModelLifecycleContext
result = context.next()

ModelLifecycleContext.finish

ModelLifecycleContext.finish(result: ResultT) -> Finish[ResultT]

Return a final value without advancing to later execution.

Parameters and example

Parameter	Type	Default
`result`	`ResultT`	Required

from harnest.lifecycle import ModelLifecycleContext

context: ModelLifecycleContext
result = context.finish(result=...)

ModelMessage

One provider-independent text message in a model call.

ModelMessage(role: str, text: str) -> None

Fields (— means no declared default):

Field	Type	Default
`role`	`str`	`—`
`text`	`str`	`—`

Parameters and example

Parameter	Type	Default
`role`	`str`	Required
`text`	`str`	Required

from harnest.lifecycle import ModelMessage

message = ModelMessage(role=..., text=...)

Continue a lifecycle chain, optionally with a replacement value.

Next(value: ValueT | _Unchanged = UNCHANGED) -> None

Fields (— means no declared default):

Field	Type	Default
`value`	`ValueT \| _Unchanged`	`UNCHANGED`

Parameters and example

Parameter	Type	Default
`value`	`ValueT \| _Unchanged`	`UNCHANGED`

from harnest.lifecycle import Next

next_value = Next()

Next.replaces

Report whether the next stage should receive a replacement value.

Example

from harnest.lifecycle import Next

next_value: Next
value = next_value.replaces

TransitionContext

Give typed lifecycle contexts one consistent control-flow vocabulary.

TransitionContext()

Example

from harnest.lifecycle import TransitionContext

context = TransitionContext()

TransitionContext.next

TransitionContext.next(replacement: Any = UNCHANGED) -> Next[Any]

Continue normally or pass a replacement to the next interceptor.

Parameters and example

Parameter	Type	Default
`replacement`	`Any`	`UNCHANGED`

from harnest.lifecycle import TransitionContext

context: TransitionContext
result = context.next()

TransitionContext.finish

TransitionContext.finish(result: ResultT) -> Finish[ResultT]

Return a final value without advancing to later execution.

Parameters and example

Parameter	Type	Default
`result`	`ResultT`	Required

from harnest.lifecycle import TransitionContext

context: TransitionContext
result = context.finish(result=...)

adk_plugin

adk_plugin(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.adk_plugin
def plugin():
    ...  # return an ADK BasePlugin

agent

Name invocation hooks by their agent lifecycle responsibility.

Example

from harnest import lifecycle

agent.before

Preserve authored hook signatures while attaching lifecycle metadata.

agent.before(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.agent.before
async def check_request(context, request):
    return context.next()

agent.after

Preserve authored hook signatures while attaching lifecycle metadata.

agent.after(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.agent.after
async def record_result(context, result):
    return context.next()

agent.on_event

Preserve authored hook signatures while attaching lifecycle metadata.

agent.on_event(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.agent.on_event
async def observe_event(context, event):
    return context.next()

agent.on_error

Preserve authored hook signatures while attaching lifecycle metadata.

agent.on_error(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.agent.on_error
async def report_failure(context, error):
    return context.next()

authenticate

Preserve authored hook signatures while attaching lifecycle metadata.

authenticate(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.authenticate
async def authenticate(connection, principal):
    ...

coverage

Build conservative defaults and apply adapter-observed coverage limits.

coverage(framework: str, mode: str, *, overrides: Mapping[str, CoverageLevel] | None = None) -> LifecycleCoverage

Parameters and example

Parameter	Type	Default
`framework`	`str`	Required
`mode`	`str`	Required
`overrides`	`Mapping[str, CoverageLevel] \| None`	`None`

from harnest import lifecycle

coverage = lifecycle.coverage(framework=..., mode=...)

credential_provider

Preserve authored hook signatures while attaching lifecycle metadata.

credential_provider(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.credential_provider
def credential_provider():
    ...

eval_capture_options

Return capture options attached by @lifecycle.evals.capture.

eval_capture_options(function: Any) -> EvalCaptureOptions | None

Parameters and example

Parameter	Type	Default
`function`	`Any`	Required

from harnest import lifecycle

options = lifecycle.eval_capture_options(function=...)

evals

Production eval improvement hooks, valid only in lifecycle/evals_improvement.py.

Example

from harnest import lifecycle

evals.capture

evals.capture(*, store: str, sample_rate: float | None = None) -> Callable[[_Hook], _Hook]

Select finished invocations for structure-only or team-redacted capture. store names a @lifecycle.storage.custom provider implementing harnest.eval_capture.CaptureStore. sample_rate additionally captures that fraction of invocations the rule rejects. Content is omitted unless @lifecycle.evals.redact is registered. That hook takes and returns one CaptureRecord; any other return value drops the capture. Harnest does not detect personal data.

Parameters and example

Parameter	Type	Default
`store`	`str`	Required
`sample_rate`	`float \| None`	`None`

from harnest import lifecycle

@lifecycle.evals.capture(store="captures", sample_rate=0.05)
def should_capture(event) -> bool:
    ...

evals.redact

Preserve authored hook signatures while attaching lifecycle metadata.

evals.redact(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.evals.redact
def redact_capture(record):
    ...

http

Group server middleware hooks under the HTTP lifecycle boundary.

Example

from harnest import lifecycle

http.scope

Preserve authored hook signatures while attaching lifecycle metadata.

http.scope(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.http.scope
def request_scope(context, request):
    ...  # return an async context manager

http.before

Preserve authored hook signatures while attaching lifecycle metadata.

http.before(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.http.before
async def check_http_request(context, request):
    return context.next()

http.after

Preserve authored hook signatures while attaching lifecycle metadata.

http.after(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.http.after
def response(context, head):
    return context.next()

http.on_error

Preserve authored hook signatures while attaching lifecycle metadata.

http.on_error(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.http.on_error
async def http_failed(context, error):
    return context.next()

http.agent_principal

Preserve authored hook signatures while attaching lifecycle metadata.

http.agent_principal(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle
from harnest.agent import AgentRuntimePrincipal
from harnest.auth import AuthPrincipal, ConnectionContext

@lifecycle.http.agent_principal
def grant(
    connection: ConnectionContext,
    principal: AuthPrincipal,
) -> AgentRuntimePrincipal:
    permissions = set()
    if principal.claims.get("can_refund") is True:
        permissions.add("refunds.write")
    return AgentRuntimePrincipal.create(permissions=permissions)

http_routes

Preserve authored hook signatures while attaching lifecycle metadata.

http_routes(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.http_routes
def http_routes(agent):
    ...  # return a FastAPI router

langgraph_middleware

langgraph_middleware(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.langgraph_middleware
def middleware():
    ...

lifecycle_coverage

Build conservative defaults and apply adapter-observed coverage limits.

lifecycle_coverage(
    framework: str,
    mode: str,
    *,
    overrides: Mapping[str, CoverageLevel] | None = None,
) -> LifecycleCoverage

Parameters and example

Parameter	Type	Default
`framework`	`str`	Required
`mode`	`str`	Required
`overrides`	`Mapping[str, CoverageLevel] \| None`	`None`

from harnest import lifecycle

coverage = lifecycle.lifecycle_coverage(framework=..., mode=...)

mcp

Group managed remote-tool hooks separately from transport ownership.

Example

from harnest import lifecycle

mcp.before

Preserve authored hook signatures while attaching lifecycle metadata.

mcp.before(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.mcp.before
async def check_remote_call(context, request):
    return context.next()

mcp.after

Preserve authored hook signatures while attaching lifecycle metadata.

mcp.after(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.mcp.after
async def check_remote_result(context, result):
    return context.next()

mcp.on_error

Preserve authored hook signatures while attaching lifecycle metadata.

mcp.on_error(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.mcp.on_error
async def remote_failed(context, error):
    return context.next()

model

Group provider-neutral model hooks under one authoring namespace.

Example

from harnest import lifecycle

model.before

Preserve authored hook signatures while attaching lifecycle metadata.

model.before(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.model.before
async def check_prompt(context, request):
    return context.next()

model.after

Preserve authored hook signatures while attaching lifecycle metadata.

model.after(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.model.after
async def check_reply(context, response):
    return context.next()

model.on_error

Preserve authored hook signatures while attaching lifecycle metadata.

model.on_error(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.model.on_error
async def model_failed(context, error):
    return context.next()

output_policy

Preserve authored hook signatures while attaching lifecycle metadata.

output_policy(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.output_policy
def output_policy():
    ...

registrations_for

Return every role attached to a possibly shared storage factory.

registrations_for(function: Any) -> tuple[LifecycleRegistration, ...]

Parameters and example

Parameter	Type	Default
`function`	`Any`	Required

from harnest import lifecycle

result = lifecycle.registrations_for(function=...)

resource

Preserve authored hook signatures while attaching lifecycle metadata.

resource(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.resource
async def catalog():
    ...

session

Observe committed sessions without participating in persistence.

Example

from harnest import lifecycle

session.created

Preserve authored hook signatures while attaching lifecycle metadata.

session.created(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.session.created
async def session_created(info, session):
    ...

skills

Example

from harnest import lifecycle

skills.source

skills.source(name: str, *, order: int = 0) -> Callable[[_Hook], _Hook]

Declare one named provider queried only during managed execution.

Parameters and example

Parameter	Type	Default	Description
`name`	`str`	Required	Name of the runtime skill catalog.
`order`	`int`	`0`	—

from harnest import lifecycle

@lifecycle.skills.source("wex")
def wex():
    ...

storage

Namespace storage roles while retaining distributed file discovery.

Example

from harnest import lifecycle

storage.sessions

Preserve authored hook signatures while attaching lifecycle metadata.

storage.sessions(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.storage.sessions
def sessions():
    ...  # return a session store

storage.checkpoints

Preserve authored hook signatures while attaching lifecycle metadata.

storage.checkpoints(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.storage.checkpoints
def checkpoints():
    ...  # return a checkpoint store

storage.tasks

Preserve authored hook signatures while attaching lifecycle metadata.

storage.tasks(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.storage.tasks
def task_storage():
    ...  # return a task store

storage.cron

Preserve authored hook signatures while attaching lifecycle metadata.

storage.cron(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.storage.cron
def cron_storage():
    ...  # return a cron store

storage.memory

Preserve authored hook signatures while attaching lifecycle metadata.

storage.memory(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.storage.memory
def memory():
    ...  # return a memory store

storage.assets

storage.assets(name: str, *, order: int = 0) -> Callable[[_Hook], _Hook]

Declare one named asset authority assembled into the registry.

Parameters and example

Parameter	Type	Default	Description
`name`	`str`	Required	Asset store name.
`order`	`int`	`0`	—

from harnest import lifecycle

@lifecycle.storage.assets("media")
def media_assets():
    ...

storage.custom

storage.custom(name: str, *, order: int = 0) -> Callable[[_Hook], _Hook]

Declare lifecycle-owned application storage under an explicit name.

Parameters and example

Parameter	Type	Default	Description
`name`	`str`	Required	Name the store is looked up by.
`order`	`int`	`0`	—

from harnest import lifecycle

@lifecycle.storage.custom("users")
def users():
    ...

telemetry_exporter

Preserve authored hook signatures while attaching lifecycle metadata.

telemetry_exporter(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.telemetry_exporter
def telemetry():
    ...

tool

Group portable tool interception under its first-class domain path.

Example

from harnest import lifecycle

tool.before

Preserve authored hook signatures while attaching lifecycle metadata.

tool.before(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default	Description
`function`	`_Hook \| None`	`None`	—
`order`	`int`	`0`	Lower values run first.

from harnest import lifecycle

@lifecycle.tool.before(order=-100)
async def check_tool(context, request):
    return context.next()

tool.after

Preserve authored hook signatures while attaching lifecycle metadata.

tool.after(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.tool.after
def tool_completed(context, result):
    return context.next()

tool.on_error

Preserve authored hook signatures while attaching lifecycle metadata.

tool.on_error(function: _Hook | None = None, *, order: int = 0) -> _Hook | Callable[[_Hook], _Hook]

Parameters and example

Parameter	Type	Default
`function`	`_Hook \| None`	`None`
`order`	`int`	`0`

from harnest import lifecycle

@lifecycle.tool.on_error
def tool_failed(context, error):
    return context.next()

​HTTP agent principal grants

​Hook control flow

​Exports

​CoverageLevel

​DROP_EVENT

​EvalCaptureOptions

​Finish

​LifecycleContext

​LifecycleContext.next

​LifecycleContext.finish

​LifecycleCoverage

​LifecycleCoverage.stage

​LifecycleCoverage.report

​LifecycleCoverage.with_overrides

​LifecycleListener

​LifecycleListener.identity

​LifecycleRegistration

​ModelCallRequest

​ModelCallResponse

​ModelLifecycleContext

​ModelLifecycleContext.next

​ModelLifecycleContext.finish

​ModelMessage

​Next

​Next.replaces

​TransitionContext

​TransitionContext.next

​TransitionContext.finish

​adk_plugin

​agent

​agent.before

​agent.after

​agent.on_event

​agent.on_error

​authenticate

​coverage

​credential_provider

​eval_capture_options

​evals

​evals.capture

​evals.redact

​http

​http.scope

​http.before

​http.after

​http.on_error

​http.agent_principal

​http_routes

​langgraph_middleware

​lifecycle_coverage

​mcp

​mcp.before

​mcp.after

​mcp.on_error

​model

​model.before

​model.after

​model.on_error

​output_policy

​registrations_for

​resource

​session

​session.created

​skills

​skills.source

​storage

​storage.sessions

​storage.checkpoints

​storage.tasks

​storage.cron

​storage.memory

​storage.assets

​storage.custom

​telemetry_exporter

​tool

​tool.before

​tool.after

​tool.on_error

HTTP agent principal grants

Hook control flow

Exports

CoverageLevel

DROP_EVENT

EvalCaptureOptions

Finish

LifecycleContext

LifecycleContext.next

LifecycleContext.finish

LifecycleCoverage

LifecycleCoverage.stage

LifecycleCoverage.report

LifecycleCoverage.with_overrides

LifecycleListener

LifecycleListener.identity

LifecycleRegistration

ModelCallRequest

ModelCallResponse

ModelLifecycleContext

ModelLifecycleContext.next

ModelLifecycleContext.finish

ModelMessage

Next

Next.replaces

TransitionContext

TransitionContext.next

TransitionContext.finish

adk_plugin

agent

agent.before

agent.after

agent.on_event

agent.on_error

authenticate

coverage

credential_provider

eval_capture_options

evals

evals.capture

evals.redact

http

http.scope

http.before

http.after

http.on_error

http.agent_principal

http_routes

langgraph_middleware

lifecycle_coverage

mcp

mcp.before

mcp.after

mcp.on_error

model

model.before

model.after

model.on_error

output_policy

registrations_for

resource

session

session.created

skills

skills.source

storage

storage.sessions

storage.checkpoints

storage.tasks

storage.cron

storage.memory

storage.assets

storage.custom

telemetry_exporter

tool

tool.before

tool.after

tool.on_error