Connections & Credentials

A connection is a small Pydantic object that describes how to reach an external service — API key, URL, region. Every node that talks to the outside world (ConnectionNode subclasses: LLMs, tools, embedders, retrievers, writers) takes a connection= argument and uses it to build its client.

from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.nodes.llms import OpenAI

llm = OpenAI(
    connection=OpenAIConnection(),  # credentials from the environment
    model="gpt-4o-mini",
)

Environment-variable conventions

Every connection field has a default_factory that reads a conventionally named environment variable, so OpenAIConnection() with no arguments "just works" when OPENAI_API_KEY is set. You can always pass values explicitly instead:

OpenAIConnection(api_key="sk-...")            # explicit (avoid hardcoding in real code)
OpenAIConnection(url="https://my-proxy/v1")   # override the endpoint, key from env

Common connections and the variables they read:

If a variable is missing, the SDK logs a warning at construction time and the field stays empty — the failure then surfaces when the node first calls the service, so check your environment early.

The full catalog (vector stores, databases, search and scraping providers, sandboxes) lives in dynamiq.connections.connections; each class documents its variables in its docstring.

Connections vs clients

ConnectionNode requires either a connection or a prebuilt client — passing neither raises 'connection' or 'client' should be specified. Use client= when you already manage an SDK client yourself (custom TLS, pooling, instrumentation) and want Dynamiq to reuse it instead of constructing its own:

from openai import OpenAI as OpenAIClient
from dynamiq.nodes.llms import OpenAI

llm = OpenAI(client=OpenAIClient(), model="gpt-4o-mini")

The ConnectionManager

Building a client per node would waste sockets and rate limits, so flows share clients through a ConnectionManager. It caches one client per unique connection (keyed by connection type plus a hash of its serialized fields), with thread-safe initialization — two nodes holding equal OpenAIConnection objects share a single client.

Every Flow creates its own manager by default; you only need to touch it to share clients across flows or to close them deterministically:

from dynamiq import Workflow
from dynamiq.connections.managers import get_connection_manager
from dynamiq.flows import Flow

with get_connection_manager() as cm:
    wf = Workflow(flow=Flow(nodes=[node_a, node_b], connection_manager=cm))
    result = wf.run(input_data={"input": "..."})
# clients are closed when the context exits

An async variant, get_async_connection_manager(), awaits client shutdown (aclose()) for async clients. Nodes loaded from YAML with postponed initialization also receive the flow's manager when their components initialize.