Non-Node Parameters

Not every parameter in a @node function is an upstream dependency. NeoGraph recognizes four kinds of parameters and resolves each one differently at runtime.

The DI surface uses typing.Annotated with FromInput / FromConfig as markers — the same FastAPI-style pattern Annotated[User, Depends(...)] uses. The primary annotation is the real type; the marker tells neograph where the value comes from.

The four kinds

Kind	Annotation	Resolved from	Example
Upstream node	Plain type (`Claims`)	Graph state, by parameter name	`decompose: Claims`
Runtime input	`Annotated[T, FromInput]`	`run(input={...})`	`topic: Annotated[str, FromInput]`
Shared resource	`Annotated[T, FromConfig]`	`config['configurable']`	`limiter: Annotated[RateLimiter, FromConfig]`
Constant	Default value	Compile-time default	`max_items: int = 10`

The framework classifies parameters at decoration time (for FromInput and FromConfig) and at construct_from_module time (for constants — because it needs the full set of decorated names to distinguish “unknown upstream” from “has a default”).

Annotated[T, FromInput]

Values injected from run(input={...}). Use this for data that varies per execution — user queries, document IDs, configuration overrides.

from typing import Annotated
from neograph import node, FromInput

@node(outputs=RawText)
def fetch(doc_id: Annotated[str, FromInput]) -> RawText:
    # doc_id comes from run(graph, input={"doc_id": "DOC-42"})
    return RawText(text=f"Contents of {doc_id}")

At runtime, run() injects all input fields into config["configurable"]. The framework reads config["configurable"]["doc_id"] and passes it to the function. If the key is absent, None is passed.

Bundling fields into a context model

When the same parameter is used by many nodes (pipeline metadata, tenant info, user context), declare a Pydantic model once and annotate the parameter with it. neograph constructs the model by pulling each of its fields from config["configurable"]:

from typing import Annotated
from pydantic import BaseModel
from neograph import node, FromInput

class RunCtx(BaseModel):
    node_id: str
    project_root: str

@node(outputs=Report)
def analyze(upstream: Claims, ctx: Annotated[RunCtx, FromInput]) -> Report:
    log(ctx.node_id, ctx.project_root)
    ...

result = run(graph, input={
    "node_id": "REQ-001",
    "project_root": "/tmp/repo",
    ...
})

Every @node that needs the context annotates one parameter with Annotated[RunCtx, FromInput] — the framework handles reconstruction. One line per node instead of three.

Annotated[T, FromConfig]

Shared resources injected via config['configurable']. Use this for objects that live across the entire pipeline — rate limiters, database connections, API clients.

from typing import Annotated
from neograph import node, FromConfig

class RateLimiter:
    def check(self) -> bool:
        return True

@node(outputs=Result)
def process(data: UpstreamData, limiter: Annotated[RateLimiter, FromConfig]) -> Result:
    if not limiter.check():
        return Result(status="rate_limited")
    return Result(status="ok")

FromConfig bundles work the same way as FromInput bundles — pass a Pydantic model as the inner type and every field is pulled from config["configurable"]:

class Shared(BaseModel):
    tenant: str
    api_key: str

@node(outputs=Result)
def process(data: Data, shared: Annotated[Shared, FromConfig]) -> Result: ...

Pass the resource when running:

graph = compile(pipeline)
result = run(graph, input={"node_id": "x"}, config={
    "configurable": {"limiter": RateLimiter()}
})

Both FromInput and FromConfig resolve from config["configurable"] at runtime. The distinction is semantic — FromInput communicates “per-execution data” while FromConfig communicates “shared infrastructure.”

Default values

Parameters with a default value that don’t match any upstream @node are treated as compile-time constants.

@node(outputs=Summary)
def summarize(claims: Claims, max_items: int = 10, verbose: bool = False) -> Summary:
    items = claims.items[:max_items]
    if verbose:
        return Summary(text=f"Processed {len(items)} of {len(claims.items)}")
    return Summary(text=f"{len(items)} items")

The default is captured at decoration time and passed to every invocation. Constants are not part of the graph topology — they don’t create edges.

Mixed parameters

A single function can use all four kinds:

from typing import Annotated
from neograph import node, FromInput, FromConfig
from pydantic import BaseModel

class Claims(BaseModel, frozen=True):
    items: list[str]

class Scores(BaseModel, frozen=True):
    ratings: dict[str, float]

class Report(BaseModel, frozen=True):
    summary: str

class RateLimiter:
    def wait(self): pass

@node(outputs=Report)
def summarize(
    claims: Claims,                        # upstream node — wired by parameter name
    scores: Scores,                        # upstream node — fan-in, second edge
    topic: Annotated[str, FromInput],                 # from run(input={"topic": "security"})
    rate_limiter: Annotated[RateLimiter, FromConfig], # from config["configurable"]["rate_limiter"]
    max_items: int = 10,                   # compile-time constant
) -> Report:
    rate_limiter.wait()
    top = sorted(scores.ratings.items(), key=lambda x: -x[1])[:max_items]
    lines = [f"{claim}: {score:.1f}" for claim, score in top]
    return Report(summary=f"Topic: {topic}\n" + "\n".join(lines))

Running the pipeline:

import sys
from neograph import construct_from_module, compile, run

pipeline = construct_from_module(sys.modules[__name__])
graph = compile(pipeline)

result = run(graph, input={
    "node_id": "review-001",
    "topic": "security audit",
}, config={
    "configurable": {"rate_limiter": RateLimiter()}
})

Resolution order

When construct_from_module processes a parameter:

If annotated Annotated[T, FromInput] — classified as runtime input.
If annotated Annotated[T, FromConfig] — classified as shared resource.
If the name matches a decorated @node in the module — classified as upstream dependency (creates an edge).
If the parameter has a default value — classified as constant.
Otherwise — ConstructError with a message listing available @node names.

This means a parameter named claims with a default value of [] will be treated as an upstream dependency if there’s a @node function called claims in the module — the upstream match takes priority over the default.