harnest.runtime

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`AgentExecutionTimeout`	Class
`AgentInfo`	Class
`AgentRuntimeError`	Class
`InvocationRequest`	Class
`InvocationResult`	Class
`NoCustomerFacingOutputError`	Class
`ResponseRequest`	Class
`RuntimeDriver`	Class
`RuntimeEvent`	Value or type alias
`SessionConflictError`	Class
`SessionMessage`	Class
`SessionRecord`	Class
`create_fastapi_app`	Function
`load_agent_card`	Function
`load_compiled_agent`	Function
`load_compiled_app`	Function
`load_compiled_application`	Function
`run_agent_message`	Function

AgentExecutionTimeout

An invocation exceeded its configured deadline.

Example

from harnest.runtime import AgentExecutionTimeout

try:
    ...
except AgentExecutionTimeout as error:
    ...

AgentInfo

Portable metadata needed by every Harnest transport.

AgentInfo(
    id: str,
    name: str,
    description: str,
    card: Mapping[str, Any],
    framework: str | None = None,
    mode: str | None = None,
    extra_endpoints: Mapping[str, str] = ...,
    input_schema: Any = None,
    output_schema: Any = None,
    lifecycle_coverage: Mapping[str, str] = ...,
    agent_principal_projection_complete: bool = False,
) -> None

Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`name`	`str`	`—`
`description`	`str`	`—`
`card`	`Mapping[str, Any]`	`—`
`framework`	`str \| None`	`None`
`mode`	`str \| None`	`None`
`extra_endpoints`	`Mapping[str, str]`	`dict()`
`input_schema`	`Any`	`None`
`output_schema`	`Any`	`None`
`lifecycle_coverage`	`Mapping[str, str]`	`dict()`
`agent_principal_projection_complete`	`bool`	`False`

Parameters and example

Parameter	Type	Default
`id`	`str`	Required
`name`	`str`	Required
`description`	`str`	Required
`card`	`Mapping[str, Any]`	Required
`framework`	`str \| None`	`None`
`mode`	`str \| None`	`None`
`extra_endpoints`	`Mapping[str, str]`	`...`
`input_schema`	`Any`	`None`
`output_schema`	`Any`	`None`
`lifecycle_coverage`	`Mapping[str, str]`	`...`
`agent_principal_projection_complete`	`bool`	`False`

from harnest.runtime import AgentInfo

info = AgentInfo(id=..., name=..., description=..., card=...)

AgentRuntimeError

A compiled artifact cannot be loaded or served safely.

Example

from harnest.runtime import AgentRuntimeError

try:
    ...
except AgentRuntimeError as error:
    ...

InvocationRequest

The complete input passed from a transport to a runtime driver.

InvocationRequest(
    input: Any,
    user_id: str,
    session_id: str,
    invocation_id: str,
    metadata: Mapping[str, Any],
    state_delta: Mapping[str, Any],
    transport: str | None = None,
    agent_principal: AgentRuntimePrincipal | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`input`	`Any`	`—`
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`invocation_id`	`str`	`—`
`metadata`	`Mapping[str, Any]`	`—`
`state_delta`	`Mapping[str, Any]`	`—`
`transport`	`str \| None`	`None`
`agent_principal`	`AgentRuntimePrincipal \| None`	`None`

Parameters and example

Parameter	Type	Default
`input`	`Any`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`invocation_id`	`str`	Required
`metadata`	`Mapping[str, Any]`	Required
`state_delta`	`Mapping[str, Any]`	Required
`transport`	`str \| None`	`None`
`agent_principal`	`AgentRuntimePrincipal \| None`	`None`

from harnest.runtime import InvocationRequest

request = InvocationRequest(
    input=...,
    user_id=...,
    session_id=...,
    invocation_id=...,
    metadata=...,
    state_delta=...,
)

InvocationResult

The normalized result of a non-streaming invocation.

InvocationResult(
    text: str,
    events: Sequence[RuntimeEvent],
    result: Any,
    session_id: str,
    metadata: Mapping[str, Any],
) -> None

Fields (— means no declared default):

Field	Type	Default
`text`	`str`	`—`
`events`	`Sequence[RuntimeEvent]`	`—`
`result`	`Any`	`—`
`session_id`	`str`	`—`
`metadata`	`Mapping[str, Any]`	`—`

Parameters and example

Parameter	Type	Default
`text`	`str`	Required
`events`	`Sequence[RuntimeEvent]`	Required
`result`	`Any`	Required
`session_id`	`str`	Required
`metadata`	`Mapping[str, Any]`	Required

from harnest.runtime import InvocationResult

result = InvocationResult(
    text=...,
    events=...,
    result=...,
    session_id=...,
    metadata=...,
)

NoCustomerFacingOutputError

Raised when a completed invocation contains no public answer or result.

Example

from harnest.runtime import NoCustomerFacingOutputError

try:
    ...
except NoCustomerFacingOutputError as error:
    ...

ResponseRequest

Strict Pydantic body accepted by default by POST /responses.

ResponseRequest(
    *,
    sessionId: str | None = None,
    stream: bool = False,
    metadata: dict[str, Any] = ...,
    input: str,
) -> None

Fields (— means no declared default):

Field	Type	Default
`session_id`	`str \| None`	`None`
`stream`	`bool`	`False`
`metadata`	`dict[str, Any]`	`dict()`
`input`	`str`	`—`

Parameters and example

Parameter	Type	Default	Description
`sessionId`	`str \| None`	`None`	—
`stream`	`bool`	`False`	Set true for text/event-stream (SSE); false returns one JSON response.
`metadata`	`dict[str, Any]`	`...`	Optional caller metadata. Does not set authentication or user identity.
`input`	`str`	Required	Non-empty message to send to the agent.

from harnest.runtime import ResponseRequest

request = ResponseRequest(input=...)

RuntimeDriver

The only interface transports require from a framework runtime.

RuntimeDriver(*args, **kwargs)

Example

from harnest.runtime import RuntimeDriver

class MyRuntimeDriver(RuntimeDriver):
    ...

RuntimeDriver.info

Return immutable metadata and structured input/output contracts.

Example

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
value = driver.info

RuntimeDriver.create_session

async RuntimeDriver.create_session(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord

Create one identity-scoped session or report a conflict.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
record = await driver.create_session(session_id=..., user_id=..., state=...)

RuntimeDriver.get_session

async RuntimeDriver.get_session(*, session_id: str, user_id: str) -> SessionRecord | None

Resolve one session without revealing another identity’s records.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
record = await driver.get_session(session_id=..., user_id=...)

RuntimeDriver.list_sessions

async RuntimeDriver.list_sessions(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]

List a bounded identity-scoped page after an optional key.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int \| None`	`None`

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
result = await driver.list_sessions(user_id=...)

RuntimeDriver.get_session_messages

async RuntimeDriver.get_session_messages(*, session_id: str, user_id: str) -> Sequence[SessionMessage] | None

Return an owned transcript while preserving native metadata.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
result = await driver.get_session_messages(session_id=..., user_id=...)

RuntimeDriver.update_session

async RuntimeDriver.update_session(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None

Apply one state delta inside the session ownership boundary.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state_delta`	`Mapping[str, Any]`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
record = await driver.update_session(session_id=..., user_id=..., state_delta=...)

RuntimeDriver.delete_session

async RuntimeDriver.delete_session(*, session_id: str, user_id: str) -> bool

Delete one owned session and report whether it existed.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
result = await driver.delete_session(session_id=..., user_id=...)

RuntimeDriver.invoke

async RuntimeDriver.invoke(request: InvocationRequest) -> InvocationResult

Run one complete non-streaming portable invocation.

Parameters and example

Parameter	Type	Default
`request`	`InvocationRequest`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
result = await driver.invoke(request=...)

RuntimeDriver.stream

RuntimeDriver.stream(request: InvocationRequest) -> AsyncIterator[RuntimeEvent]

Yield normalized events for one portable invocation.

Parameters and example

Parameter	Type	Default
`request`	`InvocationRequest`	Required

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
result = driver.stream(request=...)

RuntimeDriver.close

async RuntimeDriver.close() -> None

Release framework and application resources idempotently.

Example

from harnest.runtime import RuntimeDriver

driver: RuntimeDriver
await driver.close()

RuntimeEvent

RuntimeEvent = dict[str, Any]

Example

from harnest.runtime import RuntimeEvent

SessionConflictError

Raised by a driver when a requested session id already exists.

Example

from harnest.runtime import SessionConflictError

try:
    ...
except SessionConflictError as error:
    ...

SessionMessage

One portable transcript item with lossless framework-owned metadata.

SessionMessage(
    id: str,
    role: str,
    content: Any,
    created_at: str | None = None,
    metadata: Mapping[str, Any] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`role`	`str`	`—`
`content`	`Any`	`—`
`created_at`	`str \| None`	`None`
`metadata`	`Mapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`id`	`str`	Required
`role`	`str`	Required
`content`	`Any`	Required
`created_at`	`str \| None`	`None`
`metadata`	`Mapping[str, Any]`	`...`

from harnest.runtime import SessionMessage

message = SessionMessage(id=..., role=..., content=...)

SessionRecord

A backend-independent view of a persisted agent session.

SessionRecord(
    id: str,
    user_id: str,
    state: Mapping[str, Any],
    created_at: str | None = None,
    updated_at: str | None = None,
    metadata: Mapping[str, Any] = ...,
    application_data: Mapping[str, Any] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`user_id`	`str`	`—`
`state`	`Mapping[str, Any]`	`—`
`created_at`	`str \| None`	`None`
`updated_at`	`str \| None`	`None`
`metadata`	`Mapping[str, Any]`	`dict()`
`application_data`	`Mapping[str, Any]`	`dict()`

Parameters and example

Parameter	Type	Default
`id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required
`created_at`	`str \| None`	`None`
`updated_at`	`str \| None`	`None`
`metadata`	`Mapping[str, Any]`	`...`
`application_data`	`Mapping[str, Any]`	`...`

from harnest.runtime import SessionRecord

record = SessionRecord(id=..., user_id=..., state=...)

create_fastapi_app

Build a server with explicit policy and no UI when production assets are absent.

create_fastapi_app(
    artifact: str | Path,
    *,
    bind_host: str = '127.0.0.1',
    request_timeout: float = 300,
    max_concurrency: int = 8,
    max_request_bytes: int = 1048576,
    playground_enabled: bool = True,
    openapi_enabled: bool = True,
    live_enabled: bool = True,
    agui_enabled: bool = True,
    agent_principal_required: bool = False,
    adk_session_storage: ADKSessionStorage | None = None,
    langgraph_session_store: SessionStore | None = None,
    authenticator: Authenticator | None = None,
) -> Any

Parameters and example

Parameter	Type	Default
`artifact`	`str \| Path`	Required
`bind_host`	`str`	`'127.0.0.1'`
`request_timeout`	`float`	`300`
`max_concurrency`	`int`	`8`
`max_request_bytes`	`int`	`1048576`
`playground_enabled`	`bool`	`True`
`openapi_enabled`	`bool`	`True`
`live_enabled`	`bool`	`True`
`agui_enabled`	`bool`	`True`
`agent_principal_required`	`bool`	`False`
`adk_session_storage`	`ADKSessionStorage \| None`	`None`
`langgraph_session_store`	`SessionStore \| None`	`None`
`authenticator`	`Authenticator \| None`	`None`

from harnest.runtime import create_fastapi_app

result = create_fastapi_app(artifact=...)

load_agent_card

Load the compiled YAML or JSON Agent Card.

load_agent_card(artifact: str | Path) -> dict[str, Any]

Parameters and example

Parameter	Type	Default
`artifact`	`str \| Path`	Required

from harnest.runtime import load_agent_card

result = load_agent_card(artifact=...)

load_compiled_agent

Load the selected framework’s managed or advanced target.

load_compiled_agent(artifact: str | Path) -> Any

Parameters and example

Parameter	Type	Default
`artifact`	`str \| Path`	Required

from harnest.runtime import load_compiled_agent

result = load_compiled_agent(artifact=...)

load_compiled_app

Load the native ADK App from an ADK compiled artifact.

load_compiled_app(artifact: str | Path) -> Any

Parameters and example

Parameter	Type	Default
`artifact`	`str \| Path`	Required

from harnest.runtime import load_compiled_app

result = load_compiled_app(artifact=...)

load_compiled_application

Load the application and retain authored namespaces for lazy imports.

load_compiled_application(artifact: str | Path) -> Any

Parameters and example

Parameter	Type	Default
`artifact`	`str \| Path`	Required

from harnest.runtime import load_compiled_application

result = load_compiled_application(artifact=...)

run_agent_message

Run one request through the same driver used by the HTTP server.

async run_agent_message(
    artifact: str | Path,
    message: str,
    *,
    request_timeout: float = 300,
    agent_principal: AgentRuntimePrincipal | None = None,
) -> dict[str, Any]

Parameters and example

Parameter	Type	Default
`artifact`	`str \| Path`	Required
`message`	`str`	Required
`request_timeout`	`float`	`300`
`agent_principal`	`AgentRuntimePrincipal \| None`	`None`

from harnest.runtime import run_agent_message

result = await run_agent_message(artifact=..., message=...)

Overview

Get started

Harnest Studio

Guides

API reference

harnest.runtime

Exports

AgentExecutionTimeout

AgentInfo

AgentRuntimeError

InvocationRequest

InvocationResult

NoCustomerFacingOutputError

ResponseRequest

RuntimeDriver

RuntimeDriver.info

RuntimeDriver.create_session

RuntimeDriver.get_session

RuntimeDriver.list_sessions

RuntimeDriver.get_session_messages

RuntimeDriver.update_session

RuntimeDriver.delete_session

RuntimeDriver.invoke

RuntimeDriver.stream

RuntimeDriver.close

RuntimeEvent

SessionConflictError

SessionMessage

SessionRecord

create_fastapi_app

load_agent_card

load_compiled_agent

load_compiled_app

load_compiled_application

run_agent_message

​Exports

​AgentExecutionTimeout

​AgentInfo

​AgentRuntimeError

​InvocationRequest

​InvocationResult

​NoCustomerFacingOutputError

​ResponseRequest

​RuntimeDriver

​RuntimeDriver.info

​RuntimeDriver.create_session

​RuntimeDriver.get_session

​RuntimeDriver.list_sessions

​RuntimeDriver.get_session_messages

​RuntimeDriver.update_session

​RuntimeDriver.delete_session

​RuntimeDriver.invoke

​RuntimeDriver.stream

​RuntimeDriver.close

​RuntimeEvent

​SessionConflictError

​SessionMessage

​SessionRecord

​create_fastapi_app

​load_agent_card

​load_compiled_agent

​load_compiled_app

​load_compiled_application

​run_agent_message

Exports

AgentExecutionTimeout

AgentInfo

AgentRuntimeError

InvocationRequest

InvocationResult

NoCustomerFacingOutputError

ResponseRequest

RuntimeDriver

RuntimeDriver.info

RuntimeDriver.create_session

RuntimeDriver.get_session

RuntimeDriver.list_sessions

RuntimeDriver.get_session_messages

RuntimeDriver.update_session

RuntimeDriver.delete_session

RuntimeDriver.invoke

RuntimeDriver.stream

RuntimeDriver.close

RuntimeEvent

SessionConflictError

SessionMessage

SessionRecord

create_fastapi_app

load_agent_card

load_compiled_agent

load_compiled_app

load_compiled_application

run_agent_message