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. Inject onboarding_checklist items into the HMS prompt
3. Map soul_schema fields into the SOUL extraction logic
4. Include tool_manifest in the system prompt so the LLM knows available tools

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.