node-wire-runtime

The core of Node Wire: a framework for writing connectors, the adapters that let AI agents and applications call external systems safely. A connector declares typed actions; the runtime wraps every call in the same pipeline of validation, policy, resilience, error mapping and telemetry. A Node Wire host then exposes those actions as REST endpoints, gRPC methods and MCP tools.

Install this package to write your own connector, or as the base for the official connectors.

What it provides

Install

pip install node-wire-runtime

Cloud secret backends are optional extras. Install only the ones you use:

pip install "node-wire-runtime[aws]"    # AWS Secrets Manager (boto3)
pip install "node-wire-runtime[vault]"  # HashiCorp Vault (hvac)
pip install "node-wire-runtime[azure]"  # Azure Key Vault
pip install "node-wire-runtime[gcp]"    # GCP Secret Manager

Pre-releases are published as X.Y.ZbN; install one with pip install --pre node-wire-runtime.

The wheel is pure Python (py3-none-any) and installs on CPython 3.11, 3.12 and 3.13 on any OS and CPU.

Write a connector

A connector is a BaseConnector subclass with one @nw_action method per operation. Each action takes a Pydantic input model and returns an output model; run() validates the payload, calls the action and returns a ConnectorResponse, never raising for a failed call.

import asyncio
from typing import Literal

from pydantic import BaseModel

from node_wire_runtime import BaseConnector, EnvSecretProvider, nw_action


class GreetInput(BaseModel):
    action: Literal["greet"] = "greet"
    name: str


class GreetOutput(BaseModel):
    message: str


class GreeterConnector(BaseConnector):
    connector_id = "greeter"
    output_model = GreetOutput

    @nw_action("greet")
    async def greet(self, params: GreetInput, *, trace_id: str) -> GreetOutput:
        greeting = self.secret_provider.get_secret("GREETING")
        return GreetOutput(message=f"{greeting}, {params.name}!")


async def main() -> None:
    connector = GreeterConnector(secret_provider=EnvSecretProvider())

    response = await connector.run({"action": "greet", "name": "Ada"})
    print(response.success, response.data)  # True {'message': 'Hello, Ada!'}

    response = await connector.run({"action": "greet"})  # missing "name"
    print(response.success, response.error_code)  # False VALIDATION_ERROR


asyncio.run(main())

Run it with GREETING=Hello. To ship a connector as its own package, register the module under the node_wire.connectors entry point; see Build a connector.

Use an installed connector

Installed connectors are loaded by auto_register(), which imports only the ones named in NW_ALLOWED_CONNECTORS (comma-separated). With nothing listed, it loads nothing.

import asyncio
import os

os.environ["NW_ALLOWED_CONNECTORS"] = "http_generic"
os.environ["NW_HTTP_GENERIC_ALLOWED_HOSTS"] = "httpbin.org"

from node_wire_runtime import get_connector_registry
from node_wire_runtime.connector_registry import auto_register


async def main() -> None:
    auto_register()
    connector = get_connector_registry()["http_generic"]()
    response = await connector.run(
        {"action": "request", "method": "GET", "url": "https://httpbin.org/get"}
    )
    print(response.success, response.data["status_code"])  # True 200


asyncio.run(main())

This needs pip install node-wire-http. Connectors that use credentials are normally built by a Node Wire host (ConnectorFactory, the REST, gRPC or MCP server) from config/connectors.yaml; see Use a connector.

Official connectors

node-wire-connectors installs this runtime and all eight.

Stability

The 1.x public API is listed in the Public API reference. Anything not listed there, or named with a leading underscore, is internal. See the versioning policy.

Documentation

• Node Wire on GitHub: source, issues and full documentation
• Node Wire documentation
• Build a connector
• Configuration
• Error codes
• Changelog

License

Apache-2.0. See LICENSE.