How to use dependency injection
In shortGuide to using InjectQ for binding services and injecting them into node functions, tool functions, and agents via Inject[T] parameter defaults.
- 8 min read
- 13 sections
- Updated
- v0.10.0
- Markdown
Dependency injection in 10xGraph solves the problem of threading stateful services (databases, API clients, loggers) through your node and tool functions without polluting signatures or relying on globals. Any function registered as a node or tool can declare services as parameters with Inject[Type] as the default value, and the InjectQ container resolves and injects those dependencies automatically at runtime.
This guide walks you through binding services, injecting them into nodes and tools, and managing custom containers for tests and scoped execution.
Why dependency injection matters
Without DI, you would pass every service through config or store it as a global variable. Both approaches become unwieldy at scale. Config dictionaries become dumping grounds for untyped values, and globals make testing harder because services persist across test boundaries.
DI inverts the problem: services are bound once in a container, and functions declare the types they need. The framework wires them in at call time, without requiring explicit passing. Tools automatically receive tool_call_id, state, config, and emit from the runtime. Any other service comes from the container.
Built-in injectable parameters
Tool functions receive these parameters by name when they declare them. Node functions receive state and config by name; for anything else, declare an Inject[Type] default.
| Parameter | Type | Purpose |
|---|---|---|
tool_call_id |
str |
Unique identifier for the current tool execution (tool functions only) |
state |
AgentState |
Current graph state for reading or modifying context (nodes and tools) |
config |
dict |
Execution configuration (thread_id, user_id, recursion_limit, etc.) (nodes and tools) |
emit |
StreamEmitter |
Publishes progress events during long operations (tool functions only) |
generated_id |
str |
A fresh ID generated on each call from the ID generator |
checkpointer |
BaseCheckpointer |
The checkpointer bound to the compiled graph |
store |
BaseStore |
The long-term memory store, if configured |
publisher |
BasePublisher |
The event publisher for the graph |
context_manager |
BaseContextManager |
Context manager for multi-node operations |
task_manager |
BackgroundTaskManager |
Background task manager for long-running work |
Additionally, after compile(), these bindings are available for dependency injection:
| Binding | Type | Purpose |
|---|---|---|
CompiledGraph |
CompiledGraph |
The compiled graph instance itself |
StateGraph |
StateGraph |
The underlying StateGraph |
CallbackManager |
CallbackManager |
Callback manager for lifecycle hooks |
BaseIDGenerator |
BaseIDGenerator |
The ID generator |
BaseMediaStore |
BaseMediaStore |
Media storage (if provided at compile time) |
get_node(name) |
factory | Returns the node by name from the graph |
get_entry_point_node() |
factory | Returns the entry-point node |
generated_id_type |
str |
The ID type (e.g. “ulid”, “uuid”) |
You can register additional custom bindings before or after creating the StateGraph.
Step 1: Access the container and register a service
Every graph uses a global InjectQ singleton by default, unless you create and pass a custom container. To bind a service, get the container and call bind_instance():
from injectq import InjectQ
class DatabaseClient:
"""A simple database client."""
def __init__(self, dsn: str):
self.dsn = dsn
def query(self, sql: str) -> list:
# In a real app, execute SQL here
return []
# Register the client as a singleton
container = InjectQ.get_instance()
container.bind_instance(DatabaseClient, DatabaseClient("postgresql://localhost/mydb"))You can also bind plain key-value pairs:
container["api_key"] = "sk-1234..."
container["max_results"] = 10Or register a factory function that is called on each injection:
import uuid
container.bind_factory("request_id", lambda: str(uuid.uuid4()))Once bound, these services are available to every node and tool in the graph.
Step 2: Inject into a node function
Node functions receive state and config automatically. Declare any additional services with Inject[Type] as the parameter default:
from injectq import Inject
from tenxgraph.core.state import AgentState, Message
def query_database(
state: AgentState,
config: dict,
db: DatabaseClient = Inject[DatabaseClient],
) -> Message:
"""Query a database and return results as a message."""
results = db.query("SELECT * FROM users LIMIT 5")
content = f"Found {len(results)} users."
return Message.text_message(content, role="assistant")The framework automatically calls query_database(state, config, db=<resolved_instance>) at runtime. You never pass db manually. When you add this node to the graph, the DI system handles the wiring:
from tenxgraph.core.graph import StateGraph
graph = StateGraph()
graph.add_node("query", query_database)Step 3: Inject into a tool function
Tool functions work the same way. Declare tool_call_id, state, config, and emit as plain parameters (they are filled by the runtime). Declare any other services with Inject[Type]:
from injectq import Inject
from tenxgraph.core.state import AgentState, Message
from tenxgraph.core.state.message_block import ToolResultBlock
def search_products(
query: str,
limit: int = 5,
# --- framework-provided parameters ---
tool_call_id: str = "",
state: AgentState = None,
config: dict = None,
# --- DI-injected parameters ---
db: DatabaseClient = Inject[DatabaseClient],
) -> Message:
"""Search for products in the database."""
results = db.query(f"SELECT * FROM products WHERE name LIKE '%{query}%' LIMIT {limit}")
return Message.tool_message(
content=[ToolResultBlock(call_id=tool_call_id, output=str(results))],
)The LLM only sees query and limit in the tool schema. The framework-provided and DI-injected parameters are invisible to the model and resolved internally.
Step 4: Create and use a scoped container
For testing or when you need isolated bindings per graph, create a custom container and pass it to StateGraph():
from injectq import InjectQ
from tenxgraph.core.graph import StateGraph
# Create an isolated container for this graph
test_container = InjectQ()
test_container.bind_instance(
DatabaseClient,
DatabaseClient("postgresql://localhost/test_db")
)
# Pass the container to StateGraph
graph = StateGraph(container=test_container)
graph.add_node("query", query_database)
# ... rest of graph setup ...
app = graph.compile()
# This graph uses test_container, not the global singletonThis is especially useful in unit tests, where you want each test to have a fresh container with test doubles.
Step 5: Refresh injected values on each call
An Inject[Service] default is a proxy created once, when the function is defined, and it caches the first object it resolves. If the container is rebound later, or a graph with its own container runs, the function keeps the old object. Use fresh() from tenxgraph.utils.injection to resolve from the active container on each call:
from injectq import Inject
from tenxgraph.core.state import AgentState
from tenxgraph.storage.checkpointer import BaseCheckpointer
from tenxgraph.utils.injection import fresh
def my_node(
state: AgentState,
config: dict,
checkpointer: BaseCheckpointer = Inject[BaseCheckpointer],
) -> list:
# Without fresh(), this could be a checkpointer from an earlier graph
checkpointer = fresh(checkpointer)
return []fresh() returns an explicitly passed argument unchanged. If the dependency is not bound, it returns None, so optional services are handled gracefully.
Step 6: Read values from the container inside a node
When you need runtime access to container values, use InjectQ.get_instance().try_get():
from injectq import InjectQ
from tenxgraph.core.state import AgentState
def my_node(
state: AgentState,
config: dict,
) -> list:
"""Access the container to read optional values."""
container = InjectQ.get_instance()
# Returns None if not bound
api_key = container.try_get("api_key")
# Returns the provided default if not bound
max_results = container.try_get("max_results", 50)
# Use the values here
if api_key:
# Call external API
pass
return []This pattern is useful when a value is optional or you want to check its presence before using it.
Complete example
This example needs a provider extra, for example pip install "10xgraph[openai]", and OPENAI_API_KEY set. It builds a small graph with a database service, a tool that queries it, and a node that logs the request:
from injectq import Inject, InjectQ
from tenxgraph.core import Agent, StateGraph, ToolNode
from tenxgraph.core.state import AgentState, Message
from tenxgraph.core.state.message_block import ToolResultBlock
from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from tenxgraph.utils.constants import END
class UserRepository:
"""Simple user data store."""
def get_user(self, user_id: str) -> dict:
# Hardcoded for the example; would query a real database
return {"id": user_id, "name": "Alice", "plan": "pro"}
# Bind the repository to the global container
container = InjectQ.get_instance()
container.bind_instance(UserRepository, UserRepository())
# Tool that uses the injected repository
def get_user_info(
user_id: str,
tool_call_id: str = "",
repo: UserRepository = Inject[UserRepository],
) -> Message:
"""Get information about a user."""
user = repo.get_user(user_id)
return Message.tool_message(
content=[ToolResultBlock(call_id=tool_call_id, output=str(user))],
)
# Node that also uses the injected repository
def log_request(
state: AgentState,
config: dict,
repo: UserRepository = Inject[UserRepository],
) -> list:
"""Log the incoming request with user info."""
user_id = config.get("user_id", "unknown")
user = repo.get_user(user_id)
print(f"Request from: {user['name']} ({user['plan']} plan)")
return []
# Build the graph
tool_node = ToolNode([get_user_info])
agent = Agent(model="gpt-4o", tool_node=tool_node)
graph = StateGraph()
graph.add_node("log", log_request)
graph.add_node("MAIN", agent)
graph.add_node("TOOL", tool_node)
def should_use_tools(state: AgentState) -> str:
last = state.context[-1] if state.context else None
if last and last.role == "assistant" and getattr(last, "tools_calls", None):
return "TOOL"
return END
graph.add_edge("log", "MAIN")
graph.set_entry_point("log")
graph.add_conditional_edges("MAIN", should_use_tools, {"TOOL": "TOOL", END: END})
graph.add_edge("TOOL", "MAIN")
# Compile with a checkpointer
app = graph.compile(checkpointer=InMemoryCheckpointer())
# Run the graph
result = app.invoke(
{"messages": [Message.text_message("Get info for user ID user-42")]},
config={"thread_id": "di-demo", "user_id": "user-42"},
)
print(result["messages"][-1].text())The log node prints:
Request from: Alice (pro plan)Common patterns and errors
Pattern: Inject a scoped service per invocation. Use fresh() to always get a new instance from the container, rather than the cached version:
from tenxgraph.storage.store.base_store import BaseStore
from tenxgraph.utils.injection import fresh
def my_node(state, config, store: BaseStore = Inject[BaseStore]):
store = fresh(store)
# Uses the store bound to the active container
return []Error: “Missing required parameter” or “Required injectable parameter not found”. A tool or node declared a parameter with no default that neither the runtime nor the container provides. Bind the service before the graph runs, or give the parameter an Inject[...] default:
# Wrong: UserRepository is not bound
container = InjectQ.get_instance()
graph = StateGraph()
graph.add_node("query", my_node_using_repository) # Will fail if UserRepository is not bound
# Right: Bind first, then add the node
container.bind_instance(UserRepository, UserRepository())
graph = StateGraph()
graph.add_node("query", my_node_using_repository)Bindings after compile(). The container is compiled (frozen) at the end of graph.compile(). Add your own bindings before calling it, or create a new container and pass it to StateGraph(container=...).
What you learned
- Dependency injection avoids globals and reduces parameter passing in signatures.
- Tool functions automatically receive
tool_call_id,state,config, andemit; nodes receivestateandconfig. - Declare
param: Service = Inject[Service]to inject custom services into nodes and tools. - Use
container.bind_instance(Type, instance)to register a singleton orbind_factory(name, callable)for a function. - Pass
StateGraph(container=container)to use a custom, scoped container for testing. - Use
fresh()to resolve a dependency anew on each call, bypassing the cache.
Next steps
- Build a graph to see how DI fits into the full graph lifecycle.
- Use the @tool decorator to add metadata alongside injection.
- Dependency injection concept explains the why and tradeoffs in more detail.
Frequently asked questions
- Do I need to use dependency injection?
- No, but it beats passing everything through config. Use it when you want to inject stateful services like databases, clients, or loggers. For simple values, config works fine.
- What happens if I don't declare an injectable parameter?
- If you don't declare it, the framework won't pass it. Tool functions get tool_call_id, state, config, and emit automatically only if they declare them as parameters.
- Can I change the container after compile()?
- No. The container is frozen when you call compile(). Create a new container and pass it to StateGraph() if you need different bindings.