Skip to content

e2e_basic_agent

Level 4 · Integration — everything together: prompts, streaming, capabilities.

This example is four small demos in one file, each exercising a different core subsystem: composing a system prompt from sections, processing a stream of events, handling a capability error, and running an agent with a custom prompt. It's the "how do the pieces fit" example.

  • Lines: ~80
  • Prerequisites: none
  • Key concepts: PromptBuilder, ScopeSection, MemorySection, SkillsSection, StaticSection, stream events, CapabilityError

Run it

Installed via pip? Copy this example into your project with the CLI (no checkout needed):

pip install "symfonic-core[cli]"
symfonic examples add e2e_basic_agent
python -m e2e_basic_agent

Or, from a source checkout (run from the repo root):

python -m examples.e2e_basic_agent

Full code

"""E2E example -- PromptBuilder + StreamEvents + Capabilities.

Demonstrates:
1. PromptBuilder with multiple sections (scope, memory, skills)
2. Stream event processing
3. Capability timeout handling
4. Full agent runtime integration

Usage: python -m examples.e2e_basic_agent
"""

from __future__ import annotations

import asyncio

from symfonic.core import (
    AgentConfig,
    AgentGraph,
    AgentRuntime,
    BaseAgentDeps,
    ModelProvider,
)
from symfonic.core.contracts import (
    BuildResult,
    CapabilityError,
    ExtensionEvent,
    MessageEndEvent,
    TextDeltaEvent,
)
from symfonic.core.prompt import PromptBuilder
from symfonic.core.prompt.sections import (
    MemorySection,
    ScopeSection,
    SkillsSection,
    StaticSection,
)
from symfonic.core.streaming.event_processor import process_stream_events
from symfonic.core.testing import MockModelProvider


async def demo_prompt_builder() -> BuildResult:
    """Demo 1: Build a system prompt from sections."""
    builder = PromptBuilder()
    builder.add_section(ScopeSection(priority=10))
    builder.add_section(
        StaticSection(
            name="Identity",
            priority=20,
            content="You are a helpful AI assistant.",
        )
    )
    builder.add_section(MemorySection(priority=30))
    builder.add_section(SkillsSection(priority=40))

    state = {
        "tenant_id": "acme-corp",
        "sub_tenant_id": "engineering",
        "memory_context": {
            "last_topic": "code review",
            "preference": "concise",
        },
        "skills_context": ["summarize", "translate", "code-review"],
    }

    result: BuildResult = await builder.build(state)
    print(f"Prompt status: {result.status}")
    print(f"Warnings: {len(result.warnings)}")
    prompt_preview = result.content
    if len(prompt_preview) > 300:
        prompt_preview = prompt_preview[:300] + "..."
    print(f"Prompt ({len(result.content)} chars):")
    print(prompt_preview)
    return result


async def demo_stream_processing() -> None:
    """Demo 2: Process a stream of events."""

    async def mock_stream():  # type: ignore[no-untyped-def]
        yield TextDeltaEvent(text="Hello")
        yield TextDeltaEvent(text=" world")
        yield ExtensionEvent(type="vendor.custom", payload={"key": "value"})
        yield MessageEndEvent(message_id="msg_001", stop_reason="end_turn")

    events = await process_stream_events(mock_stream())
    print(f"\nStream events: {len(events)}")
    for e in events:
        print(f"  {type(e).__name__}: {e}")


async def demo_capability_timeout() -> None:
    """Demo 3: Capability timeout handling."""
    try:
        raise CapabilityError(
            capability="FirecrawlFetcher",
            cause="Timeout after 5.0s fetching https://slow-site.example",
            code="timeout",
            metadata={"url": "https://slow-site.example"},
        )
    except CapabilityError as exc:
        print(f"\nCaught capability error: {exc}")
        print(f"  Code: {exc.code}")
        print(f"  Metadata: {exc.metadata}")


async def demo_agent_with_prompt() -> None:
    """Demo 4: Full agent with PromptBuilder integration."""
    builder = PromptBuilder()
    builder.add_section(
        StaticSection(
            name="Identity",
            priority=10,
            content="You are symfonic, an AI agent framework assistant.",
        )
    )
    builder.add_section(MemorySection(priority=20))

    state = {"memory_context": {"project": "symfonic-core", "phase": "3"}}
    prompt_result = await builder.build(state)

    config = AgentConfig()
    deps = BaseAgentDeps()
    deps.register(
        ModelProvider,
        MockModelProvider(
            response=(
                "Phase 3 adds prompt composition, streaming, "
                "and external capabilities."
            )
        ),
    )

    graph = AgentGraph()
    runtime = AgentRuntime(graph=graph, deps=deps, config=config)

    result = await runtime.run(
        "What does Phase 3 add?",
        custom_system_prompt=prompt_result.content,
    )
    print(f"\nAgent response: {result['final_response']}")
    print(f"Nodes: {result['node_execution_log']}")


async def main() -> None:
    print("=== Demo 1: PromptBuilder ===")
    await demo_prompt_builder()

    print("\n=== Demo 2: Stream Processing ===")
    await demo_stream_processing()

    print("\n=== Demo 3: Capability Timeout ===")
    await demo_capability_timeout()

    print("\n=== Demo 4: Agent + Prompt ===")
    await demo_agent_with_prompt()

    print("\nAll demos complete!")


if __name__ == "__main__":
    asyncio.run(main())

Step by step

Demo 1 — compose a prompt from sections

builder = PromptBuilder()
builder.add_section(ScopeSection(priority=10))
builder.add_section(StaticSection(name="Identity", priority=20, content="..."))
builder.add_section(MemorySection(priority=30))
builder.add_section(SkillsSection(priority=40))
result = await builder.build(state)

PromptBuilder assembles the system prompt from independent sections, each with a priority that fixes its order. Sections read from the state dict: ScopeSection uses tenant_id, MemorySection uses memory_context, SkillsSection uses skills_context. build() returns a BuildResult with the rendered content, a status, and any warnings — so a missing section degrades gracefully instead of crashing the turn. This is the composable alternative to one giant f-string prompt.

Demo 2 — process streaming events

async def mock_stream():
    yield TextDeltaEvent(text="Hello")
    yield TextDeltaEvent(text=" world")
    yield ExtensionEvent(type="vendor.custom", payload={"key": "value"})
    yield MessageEndEvent(message_id="msg_001", stop_reason="end_turn")

events = await process_stream_events(mock_stream())

The framework models streaming as a sequence of typed events: TextDeltaEvent (a token chunk), ExtensionEvent (vendor-specific payloads), MessageEndEvent (terminal, with a stop_reason). process_stream_events consumes the async generator and returns the parsed list. Real providers emit the same event types, so UI code you write against these works regardless of the backend. See Streaming Events.

Demo 3 — handle a capability error

raise CapabilityError(
    capability="FirecrawlFetcher",
    cause="Timeout after 5.0s ...",
    code="timeout",
    metadata={"url": "..."},
)

External capabilities (web fetch, search, custom APIs) fail in structured ways. CapabilityError carries a machine-readable code ("timeout") and a metadata dict alongside the human cause, so your handler can branch on the failure mode — retry a timeout, surface an auth error — instead of parsing strings. See Capabilities.

Demo 4 — run with a custom prompt

prompt_result = await builder.build(state)
result = await runtime.run(
    "What does Phase 3 add?",
    custom_system_prompt=prompt_result.content,
)

The payoff: the prompt built in Demo 1 is passed straight into runtime.run via custom_system_prompt. The composed sections become the system prompt the model sees. This closes the loop — sections → built prompt → live agent turn.

What to try next

See also