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,
)

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,
) -> 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

    # 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

        if self._session_manager:
            self._session_manager.close(session, SessionStatus.COMPLETED)
        else:
            session.close(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)
    await self._extract_memory(prompt, response.content, 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
        if self._session_manager:
            self._session_manager.close(session, SessionStatus.COMPLETED)
        else:
            session.close(SessionStatus.COMPLETED)
        self._sessions.append(session)
        self._emit_session_event(session)

arcana.agents.registry.AgentRegistry

AgentRegistry(base_dir=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) -> None:
    self._base = base_dir or _default_base()

create

create(
    name,
    card,
    model_connection_id,
    *,
    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_connection_id: UUID,
    *,
    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_connection_id=model_connection_id,
        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 AgentRecord.model_validate_json(path.read_text())

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 = AgentRecord.model_validate_json(agent_json.read_text())
            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,
    model,
    *,
    memory=None,
    session_manager=None,
    soul=None,
)

Reconstruct a runtime Agent from a stored record, gateway, and model routing key.

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

def build_runtime(
    self,
    record: AgentRecord,
    gateway: ModelGateway,
    model: str,
    *,
    memory: MemoryAdapter | None = None,
    session_manager: SessionManager | None = None,
    soul: str | None = None,
) -> RuntimeAgent:
    """Reconstruct a runtime Agent from a stored record, gateway, and model routing key."""
    return RuntimeAgent(
        id=record.id,
        name=record.name,
        card=record.card,
        gateway=gateway,
        model=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,
    )

arcana.agents.session_manager.SessionManager

SessionManager(base_dir=None)

Manages agent sessions on disk.

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

Phase 1a: stateless — no memory extraction on close. Sessions are written to disk for audit and future memory wiring. Memory extraction lands in Phase 1b.

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

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

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.

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."""
    session.close(status)
    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)