Skip to content

symfonic.agent.fastapi.memory_create_router

memory_create_router

FastAPI router for memory-create endpoints (v6.1.4).

Five POST /memories/{layer} endpoints, one per memory layer, exposing each SDK layer's write path over HTTP so external UIs can create memory records without reaching into the Python SDK.

Design notes

  • Each layer has a distinct payload shape (semantic = label+properties, episodic = content+action_type, procedural = steps+trigger, working = ttl, prospective = trigger+task), so we ship five layer-specific endpoints rather than one generic POST /memories with a discriminated union. That matches the existing PATCH /memories/{id} + DELETE /memories/{id} idiom and keeps OpenAPI readable.
  • Request models set extra="forbid" so typos do not silently no-op, and stray tenant_id fields in the body are rejected (422) — the tenant is always derived from the X-Tenant-ID header.
  • Budget tracking for writes is deferred to a future v6.x release; writes are cheap and budget fan-out happens on LLM calls only.
  • Every create emits an AuditLogEntry via the existing emit_audit_event hook (memory.create.{layer}).

Request / response models live in memory_create_models.py. Per- layer handler bodies live in memory_create_handlers.py. This module is a thin APIRouter registration shim to keep routing concerns isolated from business logic and stay under the 300-LOC cap.

create_memory_create_router 

create_memory_create_router(
    agent: SymfonicAgent,
) -> APIRouter

Create the POST /memories/{layer} sub-router.

Mounted from create_agent_router with the same /api/v1 prefix.

Source code in src/symfonic/agent/fastapi/memory_create_router.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def create_memory_create_router(agent: SymfonicAgent) -> APIRouter:
    """Create the ``POST /memories/{layer}`` sub-router.

    Mounted from ``create_agent_router`` with the same ``/api/v1`` prefix.
    """
    router = APIRouter(tags=["memory-create"])

    @router.post("/memories/semantic", status_code=201,
                 response_model=MemoryCreateResponse)
    async def create_semantic(
        body: SemanticCreateRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> MemoryCreateResponse:
        """Create a permanent fact node. Mirrors ``SemanticLayer.store_fact``."""
        return await handle_create_semantic(agent, body, raw_request, scope)

    @router.post("/memories/episodic", status_code=201,
                 response_model=MemoryCreateResponse)
    async def create_episodic(
        body: EpisodicCreateRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> MemoryCreateResponse:
        """Create a narrative event. Mirrors ``EpisodicLayer.store_event``."""
        return await handle_create_episodic(agent, body, raw_request, scope)

    @router.post("/memories/procedural", status_code=201,
                 response_model=MemoryCreateResponse)
    async def create_procedural(
        body: ProceduralCreateRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> MemoryCreateResponse:
        """Create a skill/workflow. Mirrors ``ProceduralLayer.store_skill``."""
        return await handle_create_procedural(agent, body, raw_request, scope)

    @router.post("/memories/working", status_code=201,
                 response_model=MemoryCreateResponse)
    async def create_working(
        body: WorkingCreateRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> MemoryCreateResponse:
        """Push an entry to the working-memory buffer (FIFO)."""
        return await handle_create_working(agent, body, raw_request, scope)

    @router.post("/memories/prospective", status_code=201,
                 response_model=MemoryCreateResponse)
    async def create_prospective(
        body: ProspectiveCreateRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> MemoryCreateResponse:
        """Create a trigger/reminder. Mirrors ``ProspectiveLayer.set_trigger``."""
        return await handle_create_prospective(agent, body, raw_request, scope)

    return router