Level 4: Plugins and Domain Customization

Plugins are how you inject business-specific behavior into symfonic without modifying the framework. This level covers DomainTemplate configuration, the BaseDomainPlugin protocol, guardrails, and SOUL schema integration.

Prerequisites

• Completed Level 3
• A working agent with memory

What is a DomainTemplate?

A DomainTemplate is a frozen Pydantic model that declares what your domain needs. The framework stays generic; all business-specific configuration lives in the template.

from symfonic.agent.domain import DomainTemplate, MonitoringBlueprint

DomainTemplate Fields

Field	Type	Purpose
name	str	Human-readable domain identifier
description	str	Domain directives rendered into both HMS and JIT prompts (v7.0.2, default empty)
required_labels	list[str]	Graph labels the domain expects to exist
onboarding_checklist	list[str]	Items to collect during initial conversations
soul_schema	dict[str, str]	Schema for the SOUL label (field name to type)
tool_manifest	list[str]	Short tool descriptions injected into system prompt
tool_catalog	dict	Full JSON schemas for JIT tool hydration
tool_trigger_keywords	dict[str, list[str]]	Per-tool keyword gating for in-prompt manifest and v7.0.1 wire-level routing (v6.1.8, default empty)
identifier_keys	list[str]	Extra keys flagged by the fabrication detector (v7.0, default empty)
core_preferences	list[CorePreference]	Guardrails consulted by MetacognitiveMiddleware (priority 10 = hard refusal; 5-7 = soft guidance)
monitoring	MonitoringBlueprint	Anomaly detection thresholds and intervals

description caps. Rendered into both prompts. Truncated per FrameworkConfig.domain_description_max_chars (default (2000, 400) for (full_hms, jit)) with a single WARNING log per agent lifetime. Raise the cap via domain_description_max_chars=(5000, 800) if needed — the JIT prompt will grow in proportion.

Creating a Custom Domain

Here is a legal assistant domain:

from symfonic.agent.domain import DomainTemplate

LEGAL_DOMAIN = DomainTemplate(
    name="legal-assistant",
    required_labels=["SOUL", "CASE_LAW", "CLIENT_PROFILE", "STATUTE"],
    onboarding_checklist=[
        "Client name and contact",
        "Jurisdiction (federal/state)",
        "Legal specialty area",
        "Active case numbers",
    ],
    soul_schema={
        "name": "str",
        "jurisdiction": "str",
        "specialty": "str",
        "communication_style": "str",
        "risk_tolerance": "str",
    },
    tool_manifest=[
        "search_case_law: Search case law database by citation or topic",
        "generate_contract: Generate a contract from a template",
        "check_statute: Look up statute text by code and section",
    ],
)

Pass it to FrameworkConfig:

from symfonic.agent.config import FrameworkConfig
from symfonic.agent.engine import SymfonicAgent

config = FrameworkConfig(domain=LEGAL_DOMAIN)
agent = SymfonicAgent(model_provider=provider, config=config)

The framework uses the template to:

1. Flag missing graph labels as pending in DiscoveryService
2. List onboarding_checklist items in the memory-extraction directive, so that answers the user volunteers are recorded under the ONBOARDING label
3. Map soul_schema fields into the SOUL extraction logic
4. Include tool_manifest in the system prompt so the LLM knows available tools

onboarding_checklist does not make the agent onboard. It reaches the model only inside the extraction directive, phrased as "when the user provides answers to the onboarding checklist, create upsert_node" — a recording instruction, not an instruction to ask. Nothing in the behavioural prompt tells the agent to gather these fields, and core has no completion signal that would tell it to stop. If you want the agent to actively collect them, author that instruction yourself — and give it a stop condition, or it will still be interviewing a fully-known user on turn 500. examples/onboarding_tui/ does this with a computed prompt block that lists only the fields still missing and removes itself once none are.

Lazy tooling and tool_manifest: When lazy_tooling=True and enable_hms_prompt=True, the engine uses the procedural memory layer to select only the 3–5 tools relevant to each query instead of injecting all tool schemas on every request. A populated tool_manifest is one of the three mandatory prerequisites for this to work. See Guide 6: Lazy Tooling for the full activation recipe.

The BaseDomainPlugin Protocol

A domain plugin is any class implementing three required methods (plus one optional contribution hook added in 7.4):

class BaseDomainPlugin:
    name: str

    async def inject_system_prompt(self, state: dict) -> str:
        """Return domain-specific system prompt text.

        Called before each LLM invocation. The state dict contains
        TENANT_ID and other context from the current scope.
        """
        ...

    async def validate_state_transition(self, action: str, context: dict) -> bool:
        """Return False to block unsafe actions.

        Called before any tool execution or state change.
        Return True to allow, False to block.
        """
        ...

    def get_domain_tools(self) -> list:
        """Return LangChain StructuredTools for this domain.

        IMPORTANT: Tools must be passed at agent construction time
        via SymfonicAgent(tools=[...]). If this method returns tools,
        load_plugin() will raise SymfonicAgentError because the
        ToolRegistry freezes after graph compilation.
        """
        ...

Contribution positions (cached vs volatile, 7.4+)

BaseDomainPlugin plugins may ALSO define an optional inject_contributions hook that returns list[PluginContribution]. Each contribution declares its cache position:

• position="cached" (default) — session-stable content. Lands in the cached L1 prefix so the Anthropic prompt cache can warm it once and reuse it on every subsequent turn.
• position="volatile" — per-turn, query-conditioned content (vector search hits, RAG snippets, live counters). Lands in L2, AFTER the cache_control breakpoint, so it never invalidates the cached prefix.

from symfonic.core.plugins import PluginContribution

class QueryAwarePlugin:
    name = "QueryAware"

    async def inject_contributions(self, state):
        # Session-stable persona — cache it.
        cached = PluginContribution(
            content="You are the query-aware support agent.",
            position="cached",
        )
        # Per-turn RAG block — volatile, sits outside the cached prefix.
        query = state.get("QUERY", "")
        snippets = await self._vector_search(query)
        volatile = PluginContribution(
            content=f"Relevant docs for: {query}\n{snippets}",
            position="volatile",
        )
        return [cached, volatile]

    async def inject_system_prompt(self, state):
        # 7.4: when inject_contributions is defined, the engine SKIPS
        # this method. Leave it as a stub for Protocol compliance.
        return ""

    async def validate_state_transition(self, action, ctx): return True
    def get_domain_tools(self): return []

Migration from inject_system_prompt

The 7.4 engine dispatches per plugin:

1. If the plugin defines inject_contributions, that method is called and its list[PluginContribution] is rendered into the prompt slots. The legacy inject_system_prompt is NOT called for this plugin (no double-emit).
2. If the plugin does NOT define inject_contributions, the engine falls back to inject_system_prompt and synthesises a single PluginContribution(content=..., position="cached") so downstream rendering uses one code path.

Every existing 7.3.x plugin keeps working unchanged. Plugins that opt into inject_contributions MUST keep inject_system_prompt as at least a return "" stub (the runtime-checkable BaseDomainPlugin Protocol requires the method name to exist; an empty body is sufficient).

Cache benefit caveats

• The cache benefit on position="volatile" requires FrameworkConfig.prompt_layer_mode="stratigraphic". On the JIT path (context_strategy="jit" or legacy jit_context=True), volatile contributions are appended to the flat prompt body and provide no cache improvement; the engine emits a one-time WARNING per agent instance the first time this is observed.
• position="kernel" is reserved for a future redesign (L0 injection) and is accepted-but-no-op in 7.4: the engine remaps to position="cached" with a one-time WARNING.

Extraction protocol layout (7.4+)

In stratigraphic mode the framework now splits the built-in extraction protocol the same way it splits plugin contributions:

• The STABLE surface (JSON schema, LABEL FIELD rules, few-shot example, domain-specific labels, constraints) lives in L1 alongside the cached plugin contributions, so it warms with the Anthropic prompt cache.
• The VOLATILE KNOWN ENTITIES hint (per-turn entity ids surfaced by the hydration pipeline) lives in L2 after the cache breakpoint, next to the memory-context block that lists the same activated nodes.

This is transparent to plugin authors: the existing MemoryExtractionSection.render(state) method still returns the joined string for legacy callers, and the new render_split(state) method returns the (stable, volatile) tuple the engine uses in stratigraphic mode. No FrameworkConfig knob: legacy mode is byte-for-byte unchanged, and stratigraphic mode always benefits from the split.

Multi-section composition

A plugin can return MULTIPLE contributions per turn. The engine groups by position, sorts by (order, registration_index) (lower order wins, stable sort), and renders each contribution's content under a ### {plugin.name} header on its first appearance per (plugin, position) group. Pass header="" to suppress the auto-header (the plugin owns its own scoping), or header="## My Heading" to use a verbatim heading.

async def inject_contributions(self, state):
    return [
        PluginContribution(content="Sectional persona", position="cached", order=10),
        PluginContribution(content="Quick reference card", position="cached", order=20),
        PluginContribution(content="Live RAG hits", position="volatile"),
    ]

Example: Store Growth Plugin

A domain plugin bundles a system-prompt contribution, action guardrails, and domain tools. This illustrative plugin advises a small online store — pricing rules, listing hygiene, and a profit calculator tool:

from symfonic.core.contracts.types import PromptState

class StoreGrowthPlugin:
    name = "Store Growth"

    def __init__(self, monthly_fixed_costs: float = 0.0) -> None:
        self._fixed_costs = monthly_fixed_costs
        self._calculator = ProfitCalculator()

    async def inject_system_prompt(self, state: PromptState) -> str:
        tenant = state.get("TENANT_ID", "store")
        return f"""You are a growth assistant for the {tenant} online store.

**Listing rules:**
- Titles: Brand + Key Feature + Product Type + Size/Color (max 200 chars)
- Keep descriptions factual; no unverifiable claims

**Pricing:**
- Target margin stays above the product's landed cost
- Prefer promotions over permanent markdowns

**Guardrails:**
- NEVER recommend pricing below cost + fees + 15% margin minimum
- NEVER suggest manipulating reviews or rankings"""

    async def validate_state_transition(self, action: str, context: dict) -> bool:
        blocked = {
            "delete_listing",
            "remove_product",
            "price_below_cost",
            "manipulate_reviews",
            "fake_orders",
        }
        return action not in blocked

    def get_domain_tools(self) -> list:
        # Returns a StructuredTool wrapping ProfitCalculator.calculate()
        ...

Real Example: Mexican Criminal Law Plugin

This plugin implements a CNPP procedure state machine with strict stage ordering. It demonstrates how validate_state_transition can enforce complex domain rules:

from enum import Enum

class EtapaProcesal(str, Enum):
    INVESTIGACION_INICIAL = "investigacion_inicial"
    INVESTIGACION_COMPLEMENTARIA = "investigacion_complementaria"
    INTERMEDIA = "intermedia"
    JUICIO_ORAL = "juicio_oral"
    EJECUCION = "ejecucion"

class PenalMexicanoPlugin:
    name = "Derecho Penal MX"

    def __init__(self, initial_stage=EtapaProcesal.INVESTIGACION_INICIAL):
        self._state_machine = ProcedureStateMachine(initial=initial_stage)

    async def inject_system_prompt(self, state) -> str:
        stage_info = self._state_machine.get_stage_info()
        return f"""Eres un asistente legal especializado en Derecho Penal Mexicano.

**Etapa Procesal Actual:** {stage_info['description']}
**Articulos CNPP Relevantes:** {stage_info['cnpp_articles']}

**Guardrails:**
- NUNCA dar consejos que impliquen evadir la justicia
- SIEMPRE recomendar la asistencia de un abogado certificado"""

    async def validate_state_transition(self, action: str, context: dict) -> bool:
        # Block ethical violations unconditionally
        blocked = {"obstruct_justice", "hide_evidence", "witness_tampering"}
        if action in blocked:
            return False

        # Validate procedural stage transitions
        if action.startswith("transition_to_"):
            target_name = action[len("transition_to_"):]
            try:
                target = EtapaProcesal(target_name)
            except ValueError:
                return False
            return self._state_machine.can_transition(target)

        return True

    def get_domain_tools(self) -> list:
        # Returns a procedure_status tool
        # See examples/plugins/penal_mexicano.py for full implementation
        ...

The full implementation is in examples/plugins/penal_mexicano.py.

Loading Plugins

There is a critical constraint: domain tools must be registered at construction time. The AgentGraph's ToolRegistry freezes after compile() (called eagerly in AgentRuntime.__init__), so load_plugin() raises SymfonicAgentError if the plugin returns tools.

The correct pattern:

from myapp.plugins import StoreGrowthPlugin

# 1. Create plugin and extract tools
plugin = StoreGrowthPlugin(monthly_fixed_costs=500.0)
domain_tools = plugin.get_domain_tools()

# 2. Pass tools at construction time
agent = SymfonicAgent(
    model_provider=provider,
    tools=domain_tools,         # tools go here -- before compile()
    config=FrameworkConfig(domain=MY_DOMAIN, enable_hms_prompt=True),
)

# 3. Load plugin for prompt injection + guardrails
agent.load_plugin(plugin)       # only registers prompt + validation hooks

Multiple plugins can be loaded. Their system prompt contributions are chained in registration order.

Guardrails: validate_action()

After loading plugins, use validate_action() to check whether an action is allowed before executing it:

# Single plugin check (all plugins must agree)
allowed = await agent.validate_action("delete_listing", {"tenant": "acme"})
if not allowed:
    print("Action blocked by domain guardrails")

# Use in tool execution flow
async def safe_execute(agent, action, context, params):
    if not await agent.validate_action(action, context):
        return {"error": f"Action '{action}' blocked by domain policy"}
    return await execute_tool(action, params)

Plugin validation errors are non-fatal -- if a plugin's validate_state_transition raises an exception, it defaults to allowing the action (fail-open for availability).

SOUL Schema Integration

The soul_schema field in DomainTemplate defines what the agent should learn about each tenant. During consolidation, the HMS extracts SOUL facts from conversations and stores them as graph nodes.

ECOMMERCE_DOMAIN = DomainTemplate(
    name="ecommerce",
    soul_schema={
        "brand_voice": "str",
        "target_audience": "str",
        "competitive_edge": "str",
        "monthly_revenue": "str",
        "primary_marketplace": "str",
    },
    # ...
)

These fields are injected into the HMS system prompt. The LLM learns to extract them from conversations:

User: "Our brand targets young professionals aged 25-35 who value quality."
Agent: (extracts and stores SOUL: target_audience = "young professionals 25-35")

Query SOUL data via label_prefix:

soul_nodes = await graph_store.query_nodes(
    scope,
    label_prefix="SOUL",
)
for node in soul_nodes:
    print(f"{node.label}: {node.content}")

Full Example: Domain + Plugin + Agent

Putting it all together:

import asyncio
from symfonic.agent.config import FrameworkConfig
from symfonic.agent.domain import DomainTemplate
from symfonic.agent.engine import SymfonicAgent
from symfonic.agent.types import FrameworkTenantScope
from symfonic.core.providers import AnthropicProvider


# 1. Define the domain
MY_DOMAIN = DomainTemplate(
    name="customer-success",
    required_labels=["SOUL", "ACCOUNT", "TICKET", "PRODUCT_FEEDBACK"],
    onboarding_checklist=[
        "Company name",
        "Account tier (free/pro/enterprise)",
        "Primary contact",
        "Key pain points",
    ],
    soul_schema={
        "company_name": "str",
        "account_tier": "str",
        "primary_contact": "str",
        "satisfaction_score": "str",
    },
    tool_manifest=[
        "search_tickets: Search support tickets by keyword or status",
        "escalate_ticket: Escalate a ticket to engineering",
        "get_account_health: Retrieve account health metrics",
    ],
)


# 2. Define the plugin
class CustomerSuccessPlugin:
    name = "Customer Success"

    async def inject_system_prompt(self, state) -> str:
        return """You are a Customer Success agent.
- Prioritize customer retention and satisfaction
- Escalate P0/P1 issues immediately
- Never promise features not on the roadmap
- Always check account health before making recommendations"""

    async def validate_state_transition(self, action, context) -> bool:
        blocked = {"delete_account", "refund_without_approval", "share_internal_data"}
        return action not in blocked

    def get_domain_tools(self) -> list:
        return []  # tools passed at construction time


# 3. Wire everything together
async def main():
    plugin = CustomerSuccessPlugin()

    agent = SymfonicAgent(
        model_provider=AnthropicProvider(),
        config=FrameworkConfig(
            domain=MY_DOMAIN,
            auto_hydrate=True,
            auto_consolidate=True,
            enable_hms_prompt=True,
        ),
    )
    agent.load_plugin(plugin)

    scope = FrameworkTenantScope(tenant_id="customer-acme")

    response = await agent.run(
        "Our company is TechCorp, we're on the enterprise tier, "
        "and our main pain point is slow API response times.",
        scope=scope,
    )
    print(response.final_response)

asyncio.run(main())

Next Steps

Your agent has domain-specific behavior and guardrails. Level 5 covers production deployment with consolidation, monitoring, and scaling.

Next: Production Deployment

See also: - Feature Catalog § Domain Plugins — every DomainTemplate field with an opt-in example. - Guide 08 Fabrication Detection — how identifier_keys feeds the fabrication lexicon.