The composer has four parallel extension points. Every one is a plain registry you extend with a single register() call at import time — no subclassing, no forks, no monkey-patching.
| Extension point | What you add | Registry function |
|---|
| Agent type | A new type: key under YAML agents: | register_builder |
| LLM provider | A new provider: string under YAML model: | register_provider |
| Built-in tool | A new builtin: name under YAML tools: | register_builtin_tool |
| Tool type | A new shape of ToolDef.custom | register_tool_resolver |
orxhestra.composer:
from orxhestra.composer import (
register_builder,
register_provider,
register_builtin_tool,
register_tool_resolver,
)
Custom agent types
Register an async build function under a YAML type: key.
from orxhestra.agents.base_agent import BaseAgent
from orxhestra.composer import register_builder
from orxhestra.composer.builders.agents import Helpers
from orxhestra.composer.schema import AgentDef, ComposeSpec
class WebResearcher(BaseAgent):
"""Minimal custom agent for illustration."""
async def astream(self, user_message, context):
yield self._event("AGENT_START")
yield self._event("AGENT_MESSAGE", text=f"Researching: {user_message}")
yield self._event("AGENT_END")
async def build_web_researcher(
name: str,
agent_def: AgentDef,
spec: ComposeSpec,
*,
helpers: Helpers,
) -> BaseAgent:
"""Resolve YAML into a WebResearcher instance.
``helpers`` carries the three cross-cutting resolvers
(resolve_tools, resolve_model, build_agent) so custom builders
don't have to know about the composer's private state.
"""
tools = await helpers.resolve_tools(agent_def)
return WebResearcher(name=name, description=agent_def.description, tools=tools)
register_builder("web_researcher", build_web_researcher)
agents:
researcher:
type: web_researcher # dispatches to build_web_researcher
description: "Web research specialist"
tools:
- search
main_agent: researcher
The build function must implement the BuildFn protocol:async def build(
name: str,
agent_def: AgentDef,
spec: ComposeSpec,
*,
helpers: Helpers,
) -> BaseAgent: ...
AgentDef field.
Custom model providers
Register any class that implements LangChain’s BaseChatModel. Reference it by name via model.provider.
from langchain_core.language_models import BaseChatModel
from orxhestra.composer import register_provider
class OllamaChat(BaseChatModel):
"""Minimal stand-in — use langchain-ollama in production."""
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
...
register_provider("ollama_local", OllamaChat)
models:
local:
provider: ollama_local # resolves to OllamaChat
name: llama3
base_url: http://localhost:11434
agents:
bot:
type: llm
model: local
instructions: You run locally.
main_agent: bot
The composer ships with 20+ lazily-loaded providers (openai, anthropic, google, groq, ollama, etc.) — you probably only need register_provider for genuinely new integrations. Any unrecognised provider string is also treated as a dotted import path, so provider: myapp.models.OllamaChat works without registration.
A “built-in” is a zero-arg factory returning one or more LangChain tools. Great for project-scoped helpers that every agent in the compose spec wants.
from langchain_core.tools import tool
from orxhestra.composer import register_builtin_tool
@tool
def redact(text: str) -> str:
"""Strip emails and phone numbers from ``text``."""
...
register_builtin_tool("redact", lambda: redact)
tools:
redactor:
builtin: redact # factory is called on resolve
agents:
privacy_aware:
type: llm
tools: [redactor]
instructions: "Redact PII before replying."
main_agent: privacy_aware
redact imports something expensive, agents that don’t reference it don’t pay the cost.
If the five built-in ToolDef shapes (function, mcp, builtin, agent, transfer) don’t fit, register a whole new type of tool. The YAML escape hatch is ToolDef.custom — an arbitrary dict with a type key that the composer dispatches to your resolver.
Resolvers can be sync or async; the composer awaits whichever shape you return.
import httpx
from langchain_core.tools import Tool
from orxhestra.composer import register_tool_resolver
async def build_webhook(config: dict) -> Tool:
"""Resolver for ``custom.type == "webhook"``.
The full ``custom`` dict (including the ``type`` key) is passed
in, so the resolver can read any config the user put alongside
``type:`` in YAML.
"""
url = config["url"]
tool_name = config.get("name", "webhook")
description = config.get("description", f"POST to {url}")
async def invoke(body: dict) -> str:
async with httpx.AsyncClient() as client:
resp = await client.post(url, json=body)
return resp.text
return Tool.from_function(
name=tool_name,
description=description,
coroutine=invoke,
)
register_tool_resolver("webhook", build_webhook)
tools:
notifier:
custom:
type: webhook # dispatches to build_webhook
url: https://example.com/notify
name: notify
description: "Send a JSON payload to the notify webhook."
agents:
alerter:
type: llm
tools: [notifier]
instructions: "Call notifier whenever anomalies are detected."
main_agent: alerter
BaseTool or a list of them — useful when one YAML entry expands into a whole tool surface (e.g. a REST client that exposes each endpoint as a separate tool).
Resolvers receive the raw custom dict unchanged — no Pydantic model. That’s deliberate: you own the validation. Raise ComposerError with a helpful message when the config is invalid.
Where to register
Registration is a module-level side effect — the composer looks at the registry state at build time, so any code path that calls Composer.from_yaml(...) must have already imported the module that calls register_*.
Two common patterns:
Import before build
Entry-point convention
Per-spec registration
import myapp.agents # runs register_builder(...)
import myapp.tools # runs register_builtin_tool(...)
from orxhestra.composer import Composer
agent = Composer.from_yaml("orx.yaml")
Ship your registrations in a register() function:def register() -> None:
from orxhestra.composer import register_builder, register_builtin_tool
from myapp.agents import build_web_researcher
from myapp.tools import redact
register_builder("web_researcher", build_web_researcher)
register_builtin_tool("redact", lambda: redact)
myapp.compose_ext.register() once at app startup — keeps the side-effectful imports explicit and easy to test. Populate the registries right before you build, so each compose has its own extension set:from orxhestra.composer import Composer, register_builder
register_builder("web_researcher", build_web_researcher)
composer = Composer(spec)
root = await composer.build()
runner = await composer.build_runner(root)
Composer.build(), build_runner(), and build_server() are the public pair-points for callers that need to mutate spec or prep registries between construction and build.
Inspection APIs
Each registry exposes a read-only listing so you can verify what’s plugged in:
from orxhestra.composer.builders.agents import registered_types as agent_types
from orxhestra.composer.builders.tools import registered_tool_resolvers
print(agent_types())
# ['a2a', 'llm', 'loop', 'parallel', 'react', 'sequential', 'web_researcher']
print(registered_tool_resolvers())
# ['webhook']
type: or custom.type:, the message enumerates every registered name so you see exactly what’s legal.
See also
