Agent

arcana.agents.agent.Agent

Agent(
    name,
    card,
    gateway,
    model,
    description="",
    modifier_cards=None,
    memory=None,
    soul=None,
    system_prompt_override=None,
    id=None,
    session_manager=None,
    extractor=None,
    min_confidence_to_store=DEFAULT_MIN_CONFIDENCE_TO_STORE,
    min_confidence_for_context=DEFAULT_MIN_CONFIDENCE_FOR_CONTEXT,
    summarise_on_close=True,
)

A configured AI agent. Assign a tarot card — get a soul.

Usage

async with ModelGateway(ConnectionStore()) as gw: agent = Agent( name="researcher", card=Card.HERMIT, gateway=gw, model="ollama/hermes-3", ) result = await agent.run("summarize advances in RAG")

Source code in packages/arcana-core/arcana/agents/agent.py

def __init__(
    self,
    name: str,
    card: Card,
    gateway: ModelGateway,
    model: str,
    description: str = "",
    modifier_cards: list[Card] | None = None,
    memory: MemoryAdapter | None = None,
    soul: str | None = None,
    system_prompt_override: str | None = None,
    id: UUID | None = None,
    session_manager: SessionManager | None = None,
    extractor: MemoryExtractor | None = None,
    min_confidence_to_store: float = DEFAULT_MIN_CONFIDENCE_TO_STORE,
    min_confidence_for_context: float = DEFAULT_MIN_CONFIDENCE_FOR_CONTEXT,
    summarise_on_close: bool = True,
) -> None:
    self.id = id or uuid4()
    self.name = name
    self.card = card
    self.modifier_cards = modifier_cards or []
    self._gateway = gateway
    self._model = model
    self.memory = memory
    self.soul = soul
    self.description = description
    self._session_manager = session_manager
    # Extraction strategy — heuristic by default (deterministic, model-free).
    self._extractor = extractor or HeuristicExtractor()
    self._min_confidence_to_store = min_confidence_to_store
    self._min_confidence_for_context = min_confidence_for_context
    self._summarise_on_close = summarise_on_close

    # Resolve config from card(s)
    registry = get_registry()
    engine = CardEngine(registry)
    self._config = engine.resolve(card, self.modifier_cards)

    # Allow full system prompt override
    self._system_prompt = system_prompt_override or self._config.system_prompt
    self._temperature = self._config.temperature

    self._sessions: list[Session] = []

card_config property

card_config

The resolved AgentConfig — temperature, memory weights, etc.

run async

run(prompt, *, session=None, context=None)

Run a single prompt. Returns the assistant's response.

Pass session to resume a prior conversation; omit to start a new one.

Source code in packages/arcana-core/arcana/agents/agent.py

async def run(
    self,
    prompt: str,
    *,
    session: Session | None = None,
    context: str | None = None,
) -> str:
    """Run a single prompt. Returns the assistant's response.

    Pass *session* to resume a prior conversation; omit to start a new one.
    """
    if session is None:
        session = (
            self._session_manager.start(self.id, SessionTrigger.USER)
            if self._session_manager
            else Session(agent_id=self.id)
        )

    session.add_message(MessageRole.USER, prompt)

    history: list[MessageParam] = [
        MessageParam(role=m.role.value, content=m.content)
        for m in session.messages
        if m.role in (MessageRole.USER, MessageRole.ASSISTANT)
    ]
    history = history[-(MAX_HISTORY_TURNS * 2) :]

    with get_tracer().start_as_current_span("session.run") as span:
        span.set_attribute("arcana.agent.name", self.name)
        span.set_attribute("arcana.card", self.card.value)
        span.set_attribute("arcana.model", self._model)
        span.set_attribute("arcana.session_id", str(session.id))

        memory_context = await self._retrieve_memory_context(prompt)
        system = self._build_system(memory_context, context)

        request = CompletionRequest(
            system=system,
            messages=history,
            temperature=self._temperature,
            metadata={"session_id": str(session.id), "agent_id": str(self.id)},
        )
        response = await self._gateway.complete(self._model, request)

        session.add_message(MessageRole.ASSISTANT, response.content)
        session.total_input_tokens = response.input_tokens
        session.total_output_tokens = response.output_tokens

        await self._extract_memory(prompt, response.content, session)
        await self._close_session(session, SessionStatus.COMPLETED)

        span.set_attribute("arcana.input_tokens", response.input_tokens)
        span.set_attribute("arcana.output_tokens", response.output_tokens)
        span.set_attribute("arcana.duration_ms", session.duration_ms)

    self._emit_session_event(session)
    self._sessions.append(session)
    return response.content

stream async

stream(prompt, *, session=None, context=None)

Stream a response token by token. Records a session with token totals.

Pass session to resume a prior conversation; omit to start a new one.

Source code in packages/arcana-core/arcana/agents/agent.py

async def stream(
    self,
    prompt: str,
    *,
    session: Session | None = None,
    context: str | None = None,
) -> AsyncGenerator[str, None]:
    """Stream a response token by token. Records a session with token totals.

    Pass *session* to resume a prior conversation; omit to start a new one.
    """
    if session is None:
        session = (
            self._session_manager.start(self.id, SessionTrigger.USER)
            if self._session_manager
            else Session(agent_id=self.id)
        )

    session.add_message(MessageRole.USER, prompt)

    history: list[MessageParam] = [
        MessageParam(role=m.role.value, content=m.content)
        for m in session.messages
        if m.role in (MessageRole.USER, MessageRole.ASSISTANT)
    ]
    history = history[-(MAX_HISTORY_TURNS * 2) :]

    memory_context = await self._retrieve_memory_context(prompt)
    system = self._build_system(memory_context, context)

    request = CompletionRequest(
        system=system,
        messages=history,
        temperature=self._temperature,
        stream=True,
        metadata={"session_id": str(session.id), "agent_id": str(self.id)},
    )

    content_parts: list[str] = []
    input_tokens = 0
    output_tokens = 0
    try:
        async for chunk in self._gateway.stream(self._model, request):
            content_parts.append(chunk.text)
            input_tokens += chunk.input_tokens
            output_tokens += chunk.output_tokens
            yield chunk.text
    finally:
        full_content = "".join(content_parts)
        session.add_message(MessageRole.ASSISTANT, full_content)
        session.total_input_tokens = input_tokens
        session.total_output_tokens = output_tokens
        await self._extract_memory(prompt, full_content, session)
        await self._close_session(session, SessionStatus.COMPLETED)
        self._sessions.append(session)
        self._emit_session_event(session)

arcana.agents.registry.AgentRegistry

AgentRegistry(base_dir=None, connections=None)

Manages agent records (types.Agent) on disk.

Each record lives at ~/.arcana/agents/{id}/agent.json. The registry does not manage runtime agents (agents.Agent) directly — use build_runtime() to reconstruct one from a stored record and an adapter.

Source code in packages/arcana-core/arcana/agents/registry.py

def __init__(self, base_dir: Path | None = None, connections: ConnectionStore | None = None) -> None:
    self._base = base_dir or _default_base()
    self._connections = connections

create

create(
    name,
    card,
    model="",
    *,
    description="",
    modifier_cards=None,
    system_prompt_override=None,
    tags=None,
)

Create a new agent record, resolve card config, and persist to disk.

Source code in packages/arcana-core/arcana/agents/registry.py

def create(
    self,
    name: str,
    card: Card,
    model: str = "",
    *,
    description: str = "",
    modifier_cards: list[Card] | None = None,
    system_prompt_override: str | None = None,
    tags: list[str] | None = None,
) -> AgentRecord:
    """Create a new agent record, resolve card config, and persist to disk."""
    card_registry = get_registry()
    engine = CardEngine(card_registry)
    config = engine.resolve(card, modifier_cards or [])

    # Prompt and temperature are denormalized: computed once here and stored
    # on the record. build_runtime() feeds them back as overrides, so agents
    # are stable across card definition changes. If a definition is ever
    # revised, existing agents will keep stale prompts until re-created or
    # explicitly edited. Acceptable for MVP given card defs are treated as final.
    record = AgentRecord(
        name=name,
        card=card,
        modifier_cards=modifier_cards or [],
        model=model,
        description=description,
        system_prompt=system_prompt_override or config.system_prompt,
        temperature=config.temperature,
        tags=tags or [],
    )
    self.save(record)
    return record

get

get(agent_id)

Return an agent record by ID, or None if not found.

Source code in packages/arcana-core/arcana/agents/registry.py

def get(self, agent_id: UUID) -> AgentRecord | None:
    """Return an agent record by ID, or None if not found."""
    path = self._agent_path(agent_id)
    if not path.exists():
        return None
    return self._load_record(path)

list

list()

Return all non-archived agent records, sorted by name.

Source code in packages/arcana-core/arcana/agents/registry.py

def list(self) -> list[AgentRecord]:
    """Return all non-archived agent records, sorted by name."""
    records: list[AgentRecord] = []
    if not self._base.exists():
        return records
    for agent_dir in self._base.iterdir():
        if not agent_dir.is_dir():
            continue
        agent_json = agent_dir / "agent.json"
        if not agent_json.exists():
            continue
        try:
            record = self._load_record(agent_json)
            if not record.is_archived:
                records.append(record)
        except Exception as exc:
            print(f"warning: skipping {agent_json}: {exc}", file=sys.stderr)
    return sorted(records, key=lambda r: r.name)

save

save(record)

Persist an agent record to disk.

Source code in packages/arcana-core/arcana/agents/registry.py

def save(self, record: AgentRecord) -> None:
    """Persist an agent record to disk."""
    agent_dir = self._base / str(record.id)
    agent_dir.mkdir(parents=True, exist_ok=True)
    (agent_dir / "agent.json").write_text(record.model_dump_json(indent=2))

delete

delete(agent_id)

Soft-delete: mark the agent as archived so list() excludes it.

Source code in packages/arcana-core/arcana/agents/registry.py

def delete(self, agent_id: UUID) -> None:
    """Soft-delete: mark the agent as archived so list() excludes it."""
    record = self.get(agent_id)
    if record is None:
        raise FileNotFoundError(f"Agent {agent_id} not found.")
    self.save(record.model_copy(update={"is_archived": True}))

build_runtime

build_runtime(
    record,
    gateway,
    *,
    memory=None,
    session_manager=None,
    soul=None,
    extractor=None,
    min_confidence_to_store=DEFAULT_MIN_CONFIDENCE_TO_STORE,
    summarise_on_close=True,
)

Reconstruct a runtime Agent from a stored record and gateway.

Source code in packages/arcana-core/arcana/agents/registry.py

def build_runtime(
    self,
    record: AgentRecord,
    gateway: ModelGateway,
    *,
    memory: MemoryAdapter | None = None,
    session_manager: SessionManager | None = None,
    soul: str | None = None,
    extractor: MemoryExtractor | None = None,
    min_confidence_to_store: float = DEFAULT_MIN_CONFIDENCE_TO_STORE,
    summarise_on_close: bool = True,
) -> RuntimeAgent:
    """Reconstruct a runtime Agent from a stored record and gateway."""
    return RuntimeAgent(
        id=record.id,
        name=record.name,
        card=record.card,
        gateway=gateway,
        model=record.model,
        description=record.description,
        modifier_cards=record.modifier_cards,
        memory=memory,
        soul=soul if soul is not None else read_soul(),
        system_prompt_override=record.system_prompt,
        session_manager=session_manager,
        extractor=extractor,
        min_confidence_to_store=min_confidence_to_store,
        summarise_on_close=summarise_on_close,
    )

build_runtime_with_memory async

build_runtime_with_memory(
    record,
    gateway,
    *,
    home,
    enabled=True,
    embedding=None,
    pools=None,
    on_degraded=None,
    memory=None,
    session_manager=None,
    soul=None,
    extraction=None,
)

Build a runtime Agent with its memory federation assembled and injected.

Memory is default-on: unless the caller opts out (enabled=False) or passes an explicit memory override, this assembles a per-agent federation via build_federation and wires it in. Returns the agent together with the federation it created (None when memory is off or overridden) so the caller can close it when the run ends — the agent stays transport-agnostic and does not own that lifecycle.

The explicit memory override is passed straight through untouched, which keeps tests and specialised callers able to inject a fake adapter.

extraction selects the memory extraction strategy: the llm strategy reuses this gateway and the record's model, and falls back to the deterministic heuristic when no model is configured.

Source code in packages/arcana-core/arcana/agents/registry.py

async def build_runtime_with_memory(
    self,
    record: AgentRecord,
    gateway: ModelGateway,
    *,
    home: Path,
    enabled: bool = True,
    embedding: EmbeddingGateway | None = None,
    pools: list[PoolConfig] | None = None,
    on_degraded: Callable[[MemoryDegradedEvent], None] | None = None,
    memory: MemoryAdapter | None = None,
    session_manager: SessionManager | None = None,
    soul: str | None = None,
    extraction: ExtractionConfig | None = None,
) -> tuple[RuntimeAgent, MemoryFederation | None]:
    """Build a runtime Agent with its memory federation assembled and injected.

    Memory is default-on: unless the caller opts out (``enabled=False``) or
    passes an explicit ``memory`` override, this assembles a per-agent
    federation via ``build_federation`` and wires it in. Returns the agent
    together with the federation it created (``None`` when memory is off or
    overridden) so the caller can close it when the run ends — the agent
    stays transport-agnostic and does not own that lifecycle.

    The explicit ``memory`` override is passed straight through untouched,
    which keeps tests and specialised callers able to inject a fake adapter.

    ``extraction`` selects the memory extraction strategy: the ``llm`` strategy
    reuses this gateway and the record's model, and falls back to the
    deterministic heuristic when no model is configured.
    """
    federation: MemoryFederation | None = None
    if memory is None and enabled:
        # Card-driven decay: the agent's card (blended with any modifiers)
        # sets the per-type half-lives retrieval ranks with; The World never
        # forgets (DecayStrategy.NONE on every type).
        config = CardEngine(get_registry()).resolve(record.card, record.modifier_cards)
        decay_profiles = resolve_decay_profiles(config.decay_config, world=record.card == Card.WORLD)
        federation = await build_federation(
            record.id,
            home=home,
            embedding=embedding,
            pools=pools,
            decay_profiles=decay_profiles,
            on_degraded=on_degraded,
        )
        memory = federation

    extraction = extraction or ExtractionConfig()
    extractor = build_extractor(extraction, gateway=gateway, model=record.model or None)

    agent = self.build_runtime(
        record,
        gateway,
        memory=memory,
        session_manager=session_manager,
        soul=soul,
        extractor=extractor,
        min_confidence_to_store=extraction.min_confidence_to_store,
        summarise_on_close=extraction.summarise_on_close,
    )
    return agent, federation

arcana.agents.session_manager.SessionManager

SessionManager(base_dir=None, *, extractor=None)

Manages agent sessions on disk.

Sessions are persisted at ~/.arcana/agents/{agent_id}/sessions/{session_id}.json.

An optional extractor drives :meth:summarise; without one, summaries fall back to a deterministic, model-free heuristic.

Source code in packages/arcana-core/arcana/agents/session_manager.py

def __init__(self, base_dir: Path | None = None, *, extractor: MemoryExtractor | None = None) -> None:
    self._base = base_dir or _default_base()
    self._extractor = extractor

start

start(agent_id, trigger=USER)

Create and return a new running session. Not persisted until close().

Source code in packages/arcana-core/arcana/agents/session_manager.py

def start(
    self,
    agent_id: UUID,
    trigger: SessionTrigger = SessionTrigger.USER,
) -> Session:
    """Create and return a new running session. Not persisted until close()."""
    return Session(agent_id=agent_id, triggered_by=trigger)

append

append(session, role, content)

Add a message to a session and return the new Message.

Source code in packages/arcana-core/arcana/agents/session_manager.py

def append(self, session: Session, role: MessageRole, content: str) -> Message:
    """Add a message to a session and return the new Message."""
    return session.add_message(role, content)

close

close(session, status=COMPLETED)

Close the session and persist it to disk (no summarisation).

Source code in packages/arcana-core/arcana/agents/session_manager.py

def close(
    self,
    session: Session,
    status: SessionStatus = SessionStatus.COMPLETED,
) -> None:
    """Close the session and persist it to disk (no summarisation)."""
    session.close(status)
    self._persist(session)

summarise async

summarise(session, *, extractor=None)

Distil the session into session.summary, persist, and return it.

Uses extractor (or the manager's own), falling back to a deterministic heuristic when neither is set or the extractor raises. Never propagates an extractor failure — summarisation is best-effort.

Source code in packages/arcana-core/arcana/agents/session_manager.py

async def summarise(
    self,
    session: Session,
    *,
    extractor: MemoryExtractor | None = None,
) -> str:
    """Distil the session into ``session.summary``, persist, and return it.

    Uses *extractor* (or the manager's own), falling back to a deterministic
    heuristic when neither is set or the extractor raises. Never propagates an
    extractor failure — summarisation is best-effort.
    """
    ext = extractor or self._extractor
    text = ""
    if ext is not None:
        try:
            text = await ext.summarise(session)
        except Exception:
            logger.warning("session summarisation failed; using heuristic", exc_info=True)
    if not text:
        text = heuristic_summary(session.messages)
    session.summary = text
    self._persist(session)
    return text

close_and_summarise async

close_and_summarise(
    session,
    status=COMPLETED,
    *,
    summarise=True,
    memory=None,
    extractor=None,
)

Close the session, optionally summarising and consolidating.

When summarise is set, distils session.summary and — given a memory adapter — writes one consolidated memory carrying it. The whole summarise/consolidate step is best-effort: a failure is logged, the session is still closed and persisted, and the run is never affected.

Source code in packages/arcana-core/arcana/agents/session_manager.py

async def close_and_summarise(
    self,
    session: Session,
    status: SessionStatus = SessionStatus.COMPLETED,
    *,
    summarise: bool = True,
    memory: MemoryAdapter | None = None,
    extractor: MemoryExtractor | None = None,
) -> None:
    """Close the session, optionally summarising and consolidating.

    When ``summarise`` is set, distils ``session.summary`` and — given a
    ``memory`` adapter — writes one consolidated memory carrying it. The whole
    summarise/consolidate step is best-effort: a failure is logged, the
    session is still closed and persisted, and the run is never affected.
    """
    session.close(status)
    if not summarise:
        self._persist(session)
        return

    try:
        ext = extractor or self._extractor
        await self.summarise(session, extractor=ext)
        if memory is not None:
            signals = ext.signals if ext is not None else ENGLISH
            entry = build_consolidated_entry(session, signals=signals)
            if entry is not None:
                await memory.write(entry)
                session.memories_extracted.append(entry.id)
                self._persist(session)
    except Exception:
        logger.warning("session close summarisation/consolidation failed", exc_info=True)
        self._persist(session)

load

load(agent_id, session_id)

Load a session from disk, or None if not found.

Source code in packages/arcana-core/arcana/agents/session_manager.py

def load(self, agent_id: UUID, session_id: UUID) -> Session | None:
    """Load a session from disk, or None if not found."""
    path = self._session_path(agent_id, session_id)
    if not path.exists():
        return None
    return Session.model_validate_json(path.read_text())

list_sessions

list_sessions(agent_id)

Return all sessions for an agent, sorted by started_at ascending.

Source code in packages/arcana-core/arcana/agents/session_manager.py

def list_sessions(self, agent_id: UUID) -> list[Session]:
    """Return all sessions for an agent, sorted by started_at ascending."""
    sessions_dir = self._base / str(agent_id) / "sessions"
    if not sessions_dir.exists():
        return []
    sessions: list[Session] = []
    for path in sessions_dir.glob("*.json"):
        try:
            sessions.append(Session.model_validate_json(path.read_text()))
        except Exception:
            pass
    return sorted(sessions, key=lambda s: s.started_at)