harnest_fused

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`FusedCLIError`	Class
`FusedChannelAdapter`	Class
`FusedChannelConfig`	Class
`FusedChannelReceiver`	Class
`FusedEventMapping`	Class
`FusedMCPClient`	Class
`FusedSetupResult`	Class
`FusedSlackReplySender`	Class
`OpenAPISpec`	Class
`slack_mention`	Function

FusedCLIError

A failed provisioning phase; previous remote changes are not rolled back.

Example

from harnest_fused import FusedCLIError

try:
    ...
except FusedCLIError as error:
    ...

FusedChannelAdapter

Normalize and reply through any Fused-connected service via one MCP client. Instantiated per binding from a FusedChannelConfig, so the platform, event shape, and reply operation are all data: this class never changes.

FusedChannelAdapter(config: FusedChannelConfig) -> None

Fields (— means no declared default):

Field	Type	Default
`platform`	`str`	`—`

Parameters and example

Parameter	Type	Default
`config`	`FusedChannelConfig`	Required

from harnest_fused import FusedChannelAdapter

adapter = FusedChannelAdapter(config=...)

FusedChannelAdapter.start

async FusedChannelAdapter.start() -> None

Initialize provider-owned clients before accepting invocations.

Example

from harnest_fused import FusedChannelAdapter

adapter: FusedChannelAdapter
await adapter.start()

FusedChannelAdapter.normalize_event

FusedChannelAdapter.normalize_event(raw: Mapping[str, Any]) -> ChannelEvent

Translate one raw provider payload using this binding’s field mapping.

Parameters and example

Parameter	Type	Default
`raw`	`Mapping[str, Any]`	Required

from harnest_fused import FusedChannelAdapter

adapter: FusedChannelAdapter
event = adapter.normalize_event(raw=...)

FusedChannelAdapter.send_reply

async FusedChannelAdapter.send_reply(reply: ChannelReply) -> Mapping[str, Any]

Call the configured Fused operation directly; never through model tool-use.

Parameters and example

Parameter	Type	Default
`reply`	`ChannelReply`	Required

from harnest_fused import FusedChannelAdapter

adapter: FusedChannelAdapter
result = await adapter.send_reply(reply=...)

FusedChannelAdapter.close

async FusedChannelAdapter.close() -> None

Release provider-owned clients after invocations have drained.

Example

from harnest_fused import FusedChannelAdapter

adapter: FusedChannelAdapter
await adapter.close()

FusedChannelConfig

Everything one Fused-backed channel binding needs, expressed only as data. client is the same generic FusedMCPClient used for outbound tool calls; reply_operation is the MCP tool this adapter calls to deliver a reply, and reply_arguments maps its argument names to either a literal string or a ChannelReply field reference spelled $field_name.

FusedChannelConfig(
    platform: str,
    installation_id: str,
    mapping: FusedEventMapping,
    client: FusedMCPClient,
    reply_operation: str,
    reply_arguments: Mapping[str, str] = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`platform`	`str`	`—`
`installation_id`	`str`	`—`
`mapping`	`FusedEventMapping`	`—`
`client`	`FusedMCPClient`	`—`
`reply_operation`	`str`	`—`
`reply_arguments`	`Mapping[str, str]`	`dict()`

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`mapping`	`FusedEventMapping`	Required
`client`	`FusedMCPClient`	Required
`reply_operation`	`str`	Required
`reply_arguments`	`Mapping[str, str]`	`...`

from harnest_fused import FusedChannelConfig

config = FusedChannelConfig(
    platform=...,
    installation_id=...,
    mapping=...,
    client=...,
    reply_operation=...,
)

FusedChannelReceiver

Own one explicit generated-SDK subscription; ACK only after durable admission.

FusedChannelReceiver(receiver: Any, worker: Any, normalize: Any) -> None

Parameters and example

Parameter	Type	Default
`receiver`	`Any`	Required
`worker`	`Any`	Required
`normalize`	`Any`	Required

from harnest_fused import FusedChannelReceiver

receiver = FusedChannelReceiver(receiver=..., worker=..., normalize=...)

FusedChannelReceiver.start

FusedChannelReceiver.start(event_name: str) -> None

Subscribe to the generated enum’s exact event value, never a wildcard.

Parameters and example

Parameter	Type	Default
`event_name`	`str`	Required

from harnest_fused import FusedChannelReceiver

receiver: FusedChannelReceiver
receiver.start(event_name=...)

FusedChannelReceiver.handle

async FusedChannelReceiver.handle(payload: Mapping[str, Any], context: Mapping[str, Any]) -> None

NACK failed persistence; rejected policy events have no durable side effects.

Parameters and example

Parameter	Type	Default
`payload`	`Mapping[str, Any]`	Required
`context`	`Mapping[str, Any]`	Required

from harnest_fused import FusedChannelReceiver

receiver: FusedChannelReceiver
await receiver.handle(payload=..., context=...)

FusedChannelReceiver.close

FusedChannelReceiver.close() -> None

Stop intake before draining workers and closing their store.

Example

from harnest_fused import FusedChannelReceiver

receiver: FusedChannelReceiver
receiver.close()

FusedEventMapping

Declarative field paths translating one service’s raw payload into a ChannelEvent. Every path is dotted and resolved against the raw provider payload. Paths, not code, are what let one adapter work for any Fused-connected service.

FusedEventMapping(
    provider_event_id: str,
    kind: str,
    sender_id: str,
    conversation_id: str,
    message_id: str,
    occurred_at: str,
    content: str | None = None,
    thread_id: str | None = None,
    delivery_id: str | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`provider_event_id`	`str`	`—`
`kind`	`str`	`—`
`sender_id`	`str`	`—`
`conversation_id`	`str`	`—`
`message_id`	`str`	`—`
`occurred_at`	`str`	`—`
`content`	`str \| None`	`None`
`thread_id`	`str \| None`	`None`
`delivery_id`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`provider_event_id`	`str`	Required
`kind`	`str`	Required
`sender_id`	`str`	Required
`conversation_id`	`str`	Required
`message_id`	`str`	Required
`occurred_at`	`str`	Required
`content`	`str \| None`	`None`
`thread_id`	`str \| None`	`None`
`delivery_id`	`str \| None`	`None`

from harnest_fused import FusedEventMapping

mapping = FusedEventMapping(
    provider_event_id=...,
    kind=...,
    sender_id=...,
    conversation_id=...,
    message_id=...,
    occurred_at=...,
)

FusedEventMapping.apply

FusedEventMapping.apply(*, platform: str, installation_id: str, raw: Mapping[str, Any]) -> ChannelEvent

Build one ChannelEvent by resolving every declared path against raw.

Parameters and example

Parameter	Type	Default
`platform`	`str`	Required
`installation_id`	`str`	Required
`raw`	`Mapping[str, Any]`	Required

from harnest_fused import FusedEventMapping

mapping: FusedEventMapping
event = mapping.apply(platform=..., installation_id=..., raw=...)

FusedMCPClient

A normal MCP client carrying an optional, offline provisioning declaration. Obtain this object from the owning Harnest API; its constructor includes runtime bindings. Fields (— means no declared default):

Field	Type	Default
`transport`	`str`	`—`
`command`	`str \| None`	`None`
`args`	`Sequence[str]`	`tuple()`
`env`	`Mapping[str, str]`	`dict()`
`url`	`str \| None`	`None`
`headers`	`Mapping[str, str]`	`dict()`
`tool_filter`	`Sequence[str] \| None`	`None`
`tool_name_prefix`	`str \| None`	`None`
`timeout_seconds`	`float`	`30`
`sse_read_timeout_seconds`	`float`	`300`
`approval`	`ApprovalPolicy \| None`	`None`
`identity`	`str \| None`	`None`
`capability_id`	`str \| None`	`None`
`lifecycle`	`MCPClientLifecycle \| None`	`None`
`cwd`	`str \| None`	`None`
`portable`	`Any`	`None`
`permission`	`str \| None`	`None`
`tool_permissions`	`Mapping[str, str]`	`dict()`
`resources`	`Sequence[str] \| None`	`None`
`prompts`	`Sequence[str] \| None`	`None`
`max_content_bytes`	`int`	`1048576`
`subscriptions`	`Sequence[MCPSubscription]`	`tuple()`
`specs`	`tuple[OpenAPISpec, ...]`	`()`
`name`	`str`	`'fused'`
`url_env`	`str`	`'HARNEST_FUSED_URL'`
`token_env`	`str`	`'HARNEST_FUSED_TOKEN'`
`webhook_attachment`	`str \| None`	`None`

Example

from harnest_fused import FusedMCPClient

client: FusedMCPClient  # returned by Harnest, not constructed directly

FusedMCPClient.from_openapi

FusedMCPClient.from_openapi(
    *specs: OpenAPISpec | str | Path,
    name: str,
    url_env: str | None = None,
    token_env: str | None = None,
    webhook_attachment: str | None = None,
    **client_options: Any,
) -> FusedMCPClient

Declare one MCP connection spanning all specs without provisioning it. Strings are shorthand for OpenAPISpec with all operations selected. client_options are the standard MCPClient.streamable_http options. Authorization always uses an execution-token environment reference. webhook_attachment names a kind: webhook registration (see FusedAdminClient.apply_webhook_config) whose inbound events this server’s resources should also surface, letting any connected service become a channel’s event source alongside its outbound tool calls.

Parameters and example

Parameter	Type	Default
`*specs`	`OpenAPISpec \| str \| Path`	Optional
`name`	`str`	Required
`url_env`	`str \| None`	`None`
`token_env`	`str \| None`	`None`
`webhook_attachment`	`str \| None`	`None`
`**client_options`	`Any`	Optional

from harnest_fused import FusedMCPClient

client = FusedMCPClient.from_openapi(name=...)

FusedMCPClient.stdio

FusedMCPClient.stdio(
    command: str,
    *args: str,
    env: Mapping[str, str] | None = None,
    tools: Sequence[str] | None = None,
    prefix: str | None = None,
    permission: str | None = None,
    tool_permissions: Mapping[str, str] | None = None,
    timeout_seconds: float = 30,
    lifecycle: MCPClientLifecycle | None = None,
    resources: Sequence[str] | None = None,
    prompts: Sequence[str] | None = None,
    max_content_bytes: int = 1048576,
    subscriptions: Sequence[MCPSubscription] = (),
) -> 'MCPClient'

Create an MCP client that starts and communicates with a subprocess.

Parameters and example

Parameter	Type	Default
`command`	`str`	Required
`*args`	`str`	Optional
`env`	`Mapping[str, str] \| None`	`None`
`tools`	`Sequence[str] \| None`	`None`
`prefix`	`str \| None`	`None`
`permission`	`str \| None`	`None`
`tool_permissions`	`Mapping[str, str] \| None`	`None`
`timeout_seconds`	`float`	`30`
`lifecycle`	`MCPClientLifecycle \| None`	`None`
`resources`	`Sequence[str] \| None`	`None`
`prompts`	`Sequence[str] \| None`	`None`
`max_content_bytes`	`int`	`1048576`
`subscriptions`	`Sequence[MCPSubscription]`	`()`

from harnest_fused import FusedMCPClient

client = FusedMCPClient.stdio(command=...)

FusedMCPClient.streamable_http

FusedMCPClient.streamable_http(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    tools: Sequence[str] | None = None,
    prefix: str | None = None,
    permission: str | None = None,
    tool_permissions: Mapping[str, str] | None = None,
    timeout_seconds: float = 30,
    sse_read_timeout_seconds: float = 300,
    lifecycle: MCPClientLifecycle | None = None,
    resources: Sequence[str] | None = None,
    prompts: Sequence[str] | None = None,
    max_content_bytes: int = 1048576,
    subscriptions: Sequence[MCPSubscription] = (),
) -> 'MCPClient'

Create an MCP client using the Streamable HTTP transport.

Parameters and example

Parameter	Type	Default
`url`	`str`	Required
`headers`	`Mapping[str, str] \| None`	`None`
`tools`	`Sequence[str] \| None`	`None`
`prefix`	`str \| None`	`None`
`permission`	`str \| None`	`None`
`tool_permissions`	`Mapping[str, str] \| None`	`None`
`timeout_seconds`	`float`	`30`
`sse_read_timeout_seconds`	`float`	`300`
`lifecycle`	`MCPClientLifecycle \| None`	`None`
`resources`	`Sequence[str] \| None`	`None`
`prompts`	`Sequence[str] \| None`	`None`
`max_content_bytes`	`int`	`1048576`
`subscriptions`	`Sequence[MCPSubscription]`	`()`

from harnest_fused import FusedMCPClient

client = FusedMCPClient.streamable_http(url=...)

FusedMCPClient.sse

FusedMCPClient.sse(
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    tools: Sequence[str] | None = None,
    prefix: str | None = None,
    permission: str | None = None,
    tool_permissions: Mapping[str, str] | None = None,
    timeout_seconds: float = 30,
    sse_read_timeout_seconds: float = 300,
    lifecycle: MCPClientLifecycle | None = None,
    resources: Sequence[str] | None = None,
    prompts: Sequence[str] | None = None,
    max_content_bytes: int = 1048576,
    subscriptions: Sequence[MCPSubscription] = (),
) -> 'MCPClient'

Create an MCP client using the legacy HTTP/SSE transport.

Parameters and example

Parameter	Type	Default
`url`	`str`	Required
`headers`	`Mapping[str, str] \| None`	`None`
`tools`	`Sequence[str] \| None`	`None`
`prefix`	`str \| None`	`None`
`permission`	`str \| None`	`None`
`tool_permissions`	`Mapping[str, str] \| None`	`None`
`timeout_seconds`	`float`	`30`
`sse_read_timeout_seconds`	`float`	`300`
`lifecycle`	`MCPClientLifecycle \| None`	`None`
`resources`	`Sequence[str] \| None`	`None`
`prompts`	`Sequence[str] \| None`	`None`
`max_content_bytes`	`int`	`1048576`
`subscriptions`	`Sequence[MCPSubscription]`	`()`

from harnest_fused import FusedMCPClient

client = FusedMCPClient.sse(url=...)

FusedMCPClient.connect

FusedMCPClient.connect(*, framework: Literal['adk', 'langgraph'] = 'langgraph') -> Any

Open a developer-owned read client using the configured lifecycle and credentials.

Parameters and example

Parameter	Type	Default
`framework`	`Literal['adk', 'langgraph']`	`'langgraph'`

from harnest_fused import FusedMCPClient

client: FusedMCPClient
result = client.connect()

FusedMCPClient.required_permissions_for

FusedMCPClient.required_permissions_for(tool_name: str) -> frozenset[str]

Combine an optional client requirement with one remote-tool requirement.

Parameters and example

Parameter	Type	Default
`tool_name`	`str`	Required

from harnest_fused import FusedMCPClient

client: FusedMCPClient
result = client.required_permissions_for(tool_name=...)

FusedMCPClient.to_adk_toolset

FusedMCPClient.to_adk_toolset() -> Any

Use native transport while preserving portable expansion and failure rules.

Example

from harnest_fused import FusedMCPClient

client: FusedMCPClient
result = client.to_adk_toolset()

FusedMCPClient.to_langgraph_connection

FusedMCPClient.to_langgraph_connection() -> dict[str, Any]

Return a LangChain MCP adapter connection without opening it.

Example

from harnest_fused import FusedMCPClient

client: FusedMCPClient
result = client.to_langgraph_connection()

FusedMCPClient.setup

FusedMCPClient.setup(
    *,
    description: str,
    bucket: str,
    version: str = '1.0.0',
    project: str | Path = '.',
    cli: str = 'fused-cli',
    timeout_seconds: float = 1260,
) -> FusedSetupResult

Explicitly import specs and provision the server using the user’s CLI. This is a developer setup operation, never a runtime lifecycle hook. The CLI displays its one-time execution token directly to the operator.

Parameters and example

Parameter	Type	Default
`description`	`str`	Required
`bucket`	`str`	Required
`version`	`str`	`'1.0.0'`
`project`	`str \| Path`	`'.'`
`cli`	`str`	`'fused-cli'`
`timeout_seconds`	`float`	`1260`

from harnest_fused import FusedMCPClient

client: FusedMCPClient
result = client.setup(description=..., bucket=...)

FusedSetupResult

A provisioned standard client and non-secret setup artifacts. The CLI hands its one-time token directly to the operator. Store it in the environment variable named by token_env before connecting this client.

FusedSetupResult(
    client: MCPClient,
    url: str,
    url_env: str,
    token_env: str,
    config_path: Path,
    receipt_directory: Path,
) -> None

Fields (— means no declared default):

Field	Type	Default
`client`	`MCPClient`	`—`
`url`	`str`	`—`
`url_env`	`str`	`—`
`token_env`	`str`	`—`
`config_path`	`Path`	`—`
`receipt_directory`	`Path`	`—`

Parameters and example

Parameter	Type	Default
`client`	`MCPClient`	Required
`url`	`str`	Required
`url_env`	`str`	Required
`token_env`	`str`	Required
`config_path`	`Path`	Required
`receipt_directory`	`Path`	Required

from harnest_fused import FusedSetupResult

result = FusedSetupResult(
    client=...,
    url=...,
    url_env=...,
    token_env=...,
    config_path=...,
    receipt_directory=...,
)

FusedSetupResult.write_client

FusedSetupResult.write_client(path: str | Path) -> Path

Create a standard client factory without overwriting authored code.

Parameters and example

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

from harnest_fused import FusedSetupResult

result: FusedSetupResult
result = result.write_client(path=...)

FusedSlackReplySender

Call the SDK-scoped Engine API, keeping provider credentials inside Fused. The operation must be the exact Slack chat.postMessage selection from the immutable app’s exported execution schema. The caller owns the HTTP client authenticated with that SDK’s execution token and an explicit user selector.

FusedSlackReplySender(client: Any, *, app_id: str, operation: str, selector: Mapping[str, str]) -> None

Parameters and example

Parameter	Type	Default
`client`	`Any`	Required
`app_id`	`str`	Required
`operation`	`str`	Required
`selector`	`Mapping[str, str]`	Required

from harnest_fused import FusedSlackReplySender

sender = FusedSlackReplySender(client=..., app_id=..., operation=..., selector=...)

FusedSlackReplySender.call

async FusedSlackReplySender.__call__(reply: ChannelReply) -> dict[str, str]

Send once, suppress broadcasts/unfurls, and check provider-level success.

Parameters and example

Parameter	Type	Default
`reply`	`ChannelReply`	Required

from harnest_fused import FusedSlackReplySender

sender: FusedSlackReplySender
result = await sender(reply=...)

OpenAPISpec

One source service; omitted operations select its entire operation surface. version supplies the provider version only when the spec omits it. auth identifies credentials already managed in the chosen Fused bucket. Source paths resolve against the project passed to setup.

OpenAPISpec(
    source: str | Path,
    name: str | None = None,
    operations: Sequence[str] | None = None,
    version: str | None = None,
    auth: Mapping[str, str] | None = None,
    scopes: Sequence[str] | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`source`	`str \| Path`	`—`
`name`	`str \| None`	`None`
`operations`	`Sequence[str] \| None`	`None`
`version`	`str \| None`	`None`
`auth`	`Mapping[str, str] \| None`	`None`
`scopes`	`Sequence[str] \| None`	`None`

Parameters and example

Parameter	Type	Default
`source`	`str \| Path`	Required
`name`	`str \| None`	`None`
`operations`	`Sequence[str] \| None`	`None`
`version`	`str \| None`	`None`
`auth`	`Mapping[str, str] \| None`	`None`
`scopes`	`Sequence[str] \| None`	`None`

from harnest_fused import OpenAPISpec

spec = OpenAPISpec(source=...)

OpenAPISpec.selection

OpenAPISpec.selection(version: str) -> dict[str, Any]

Project selection into Fused’s single authoritative MCP config shape.

Parameters and example

Parameter	Type	Default
`version`	`str`	Required

from harnest_fused import OpenAPISpec

spec: OpenAPISpec
result = spec.selection(version=...)

slack_mention

Normalize a verified Fused Slack envelope, rejecting foreign installations and bots. Call only from the authenticated generated SDK receiver. IDs in a raw public HTTP request are not proof of provider signature or connection ownership.

slack_mention(payload: Mapping[str, Any], *, installation_id: str, app_id: str) -> ChannelEvent | None

Parameters and example

Parameter	Type	Default
`payload`	`Mapping[str, Any]`	Required
`installation_id`	`str`	Required
`app_id`	`str`	Required

from harnest_fused import slack_mention

event = slack_mention(payload=..., installation_id=..., app_id=...)

Overview

Get started

Harnest Studio

Guides

API reference

harnest_fused

Exports

FusedCLIError

FusedChannelAdapter

FusedChannelAdapter.start

FusedChannelAdapter.normalize_event

FusedChannelAdapter.send_reply

FusedChannelAdapter.close

FusedChannelConfig

FusedChannelReceiver

FusedChannelReceiver.start

FusedChannelReceiver.handle

FusedChannelReceiver.close

FusedEventMapping

FusedEventMapping.apply

FusedMCPClient

FusedMCPClient.from_openapi

FusedMCPClient.stdio

FusedMCPClient.streamable_http

FusedMCPClient.sse

FusedMCPClient.connect

FusedMCPClient.required_permissions_for

FusedMCPClient.to_adk_toolset

FusedMCPClient.to_langgraph_connection

FusedMCPClient.setup

FusedSetupResult

FusedSetupResult.write_client

FusedSlackReplySender

FusedSlackReplySender.call

OpenAPISpec

OpenAPISpec.selection

slack_mention

​Exports

​FusedCLIError

​FusedChannelAdapter

​FusedChannelAdapter.start

​FusedChannelAdapter.normalize_event

​FusedChannelAdapter.send_reply

​FusedChannelAdapter.close

​FusedChannelConfig

​FusedChannelReceiver

​FusedChannelReceiver.start

​FusedChannelReceiver.handle

​FusedChannelReceiver.close

​FusedEventMapping

​FusedEventMapping.apply

​FusedMCPClient

​FusedMCPClient.from_openapi

​FusedMCPClient.stdio

​FusedMCPClient.streamable_http

​FusedMCPClient.sse

​FusedMCPClient.connect

​FusedMCPClient.required_permissions_for

​FusedMCPClient.to_adk_toolset

​FusedMCPClient.to_langgraph_connection

​FusedMCPClient.setup

​FusedSetupResult

​FusedSetupResult.write_client

​FusedSlackReplySender

​FusedSlackReplySender.call

​OpenAPISpec

​OpenAPISpec.selection

​slack_mention

Exports

FusedCLIError

FusedChannelAdapter

FusedChannelAdapter.start

FusedChannelAdapter.normalize_event

FusedChannelAdapter.send_reply

FusedChannelAdapter.close

FusedChannelConfig

FusedChannelReceiver

FusedChannelReceiver.start

FusedChannelReceiver.handle

FusedChannelReceiver.close

FusedEventMapping

FusedEventMapping.apply

FusedMCPClient

FusedMCPClient.from_openapi

FusedMCPClient.stdio

FusedMCPClient.streamable_http

FusedMCPClient.sse

FusedMCPClient.connect

FusedMCPClient.required_permissions_for

FusedMCPClient.to_adk_toolset

FusedMCPClient.to_langgraph_connection

FusedMCPClient.setup

FusedSetupResult

FusedSetupResult.write_client

FusedSlackReplySender

FusedSlackReplySender.call

OpenAPISpec

OpenAPISpec.selection

slack_mention