API Reference¶

from cognitia import Agent, AgentConfig

agent = Agent(config: AgentConfig)

async with Agent(config) as agent:
    result = await agent.query("Hello")
# cleanup called automatically

from cognitia import AgentConfig

from cognitia.agent import Result

conv = agent.conversation(session_id="optional-id")

from cognitia import tool

@tool(name="my_tool", description="Description for LLM", schema=None)
async def my_tool(param: str) -> str:
    return "result"

td = my_tool.__tool_definition__
td.name          # "my_tool"
td.description   # "Description for LLM"
td.parameters    # {"type": "object", "properties": {...}, "required": [...]}
td.handler       # async function reference
td.to_tool_spec() # ToolSpec for runtime

from cognitia.agent import Middleware

class MyMiddleware(Middleware):
    async def before_query(self, prompt: str, config: AgentConfig) -> str:
        return prompt  # can modify prompt; raise to block

    async def after_result(self, result: Result) -> Result:
        return result  # can modify result

    def get_hooks(self) -> HookRegistry | None:
        return None  # optional lifecycle hooks

from cognitia.agent import CostTracker

tracker = CostTracker(budget_usd=5.0)
tracker.total_cost_usd  # float — accumulated cost
tracker.reset()         # reset counter
# Raises BudgetExceededError when cost > budget

from cognitia.agent import SecurityGuard

guard = SecurityGuard(block_patterns=["password", "secret"])
# Blocks tool inputs containing any of the patterns via PreToolUse hook

from cognitia.runtime.types import Message

msg = Message(role="user", content="Hello")

from cognitia.runtime.types import ToolSpec

spec = ToolSpec(name="my_tool", description="...", parameters={...}, is_local=True)

from cognitia.runtime.types import RuntimeConfig

config = RuntimeConfig(
    runtime_name="thin",              # claude_sdk | thin | deepagents | cli
    model="claude-sonnet-4-20250514", # or alias
    max_iterations=6,                 # ReAct loop limit
    max_tool_calls=8,                 # tool calls per turn limit
    max_model_retries=2,              # retry on bad output
    base_url=None,                    # custom API endpoint
    feature_mode="portable",          # portable | hybrid | native_first
    required_capabilities=None,       # CapabilityRequirements
    allow_native_features=False,      # allow runtime-specific features
    native_config={},                 # runtime-specific extra config
)

from cognitia.runtime.types import TurnMetrics

from cognitia.runtime.capabilities import get_runtime_capabilities

caps = get_runtime_capabilities("claude_sdk")
caps.tier              # "full"
caps.supports_mcp      # True
caps.enabled_flags()   # frozenset({"mcp", "resume", "interrupt"})

from cognitia.runtime.capabilities import CapabilityRequirements

reqs = CapabilityRequirements(
    tier="full",
    flags=("mcp", "resume"),
)

from cognitia.runtime.types import resolve_model_name

resolve_model_name("sonnet")   # "claude-sonnet-4-20250514"
resolve_model_name("opus")     # "claude-opus-4-20250514"
resolve_model_name("haiku")    # "claude-haiku-3-20250307"
resolve_model_name("gpt-4o")   # "gpt-4o"
resolve_model_name("4o-mini")  # "gpt-4o-mini"
resolve_model_name("o3")       # "o3"
resolve_model_name("gemini")   # "gemini-2.5-pro"
resolve_model_name("r1")       # "deepseek-reasoner"
resolve_model_name(None)       # default model

class MessageStore(Protocol):
    async def save_message(self, user_id: str, topic_id: str, role: str, content: str, tool_calls: list[dict] | None = None) -> None: ...
    async def get_messages(self, user_id: str, topic_id: str, limit: int = 50) -> list[MemoryMessage]: ...
    async def count_messages(self, user_id: str, topic_id: str) -> int: ...
    async def delete_messages_before(self, user_id: str, topic_id: str, keep_last: int = 10) -> int: ...

class FactStore(Protocol):
    async def upsert_fact(self, user_id: str, key: str, value: str, source: str = "") -> None: ...
    async def get_facts(self, user_id: str) -> dict[str, str]: ...

class GoalStore(Protocol):
    async def save_goal(self, user_id: str, topic_id: str, data: dict) -> None: ...
    async def get_active_goal(self, user_id: str, topic_id: str) -> dict | None: ...

class SummaryStore(Protocol):
    async def save_summary(self, user_id: str, topic_id: str, summary: str, messages_covered: int) -> None: ...
    async def get_summary(self, user_id: str, topic_id: str) -> str | None: ...

class UserStore(Protocol):
    async def ensure_user(self, external_id: str) -> str: ...
    async def get_user_profile(self, user_id: str) -> dict | None: ...

class SessionStateStore(Protocol):
    async def save_session_state(self, user_id: str, topic_id: str, role_id: str, active_skill_ids: list[str], prompt_hash: str) -> None: ...
    async def get_session_state(self, user_id: str, topic_id: str) -> dict | None: ...

class PhaseStore(Protocol):
    async def save_phase_state(self, user_id: str, phase: str, notes: str = "") -> None: ...
    async def get_phase_state(self, user_id: str) -> dict | None: ...

class ToolEventStore(Protocol):
    async def save_tool_event(self, user_id: str, event_data: dict) -> None: ...

class WebProvider(Protocol):
    async def fetch(self, url: str) -> str: ...
    async def search(self, query: str, max_results: int = 5) -> list[SearchResult]: ...

class WebSearchProvider(Protocol):
    async def search(self, query: str, max_results: int = 5) -> list[SearchResult]: ...

class WebFetchProvider(Protocol):
    async def fetch(self, url: str) -> str: ...

@dataclass(frozen=True)
class SearchResult:
    title: str
    url: str
    snippet: str

from cognitia.hooks import HookRegistry

registry = HookRegistry()
registry.on_pre_tool_use(callback, matcher="Bash")
registry.on_post_tool_use(callback)
registry.on_stop(callback)
registry.on_user_prompt(callback)

from cognitia.hooks import registry_to_sdk_hooks

sdk_hooks = registry_to_sdk_hooks(registry)
# Pass to Claude Agent SDK options

from cognitia.observability import AgentLogger, configure_logging

configure_logging(level="info", fmt="json")
logger = AgentLogger(component="my_app")

logger.session_created(user_id="u1", topic_id="t1", role_id="coach")
logger.turn_start(user_id="u1", topic_id="t1")
logger.tool_call(tool_name="get_deposits", latency_ms=450)
logger.tool_policy_event(tool_name="Bash", allowed=False, reason="ALWAYS_DENIED")
logger.turn_complete(user_id="u1", topic_id="t1", role_id="coach", prompt_hash="abc123")

from cognitia.resilience import CircuitBreaker

cb = CircuitBreaker(failure_threshold=3, recovery_timeout_s=60)

if cb.can_execute():
    try:
        result = await call_service()
        cb.record_success()
    except Exception:
        cb.record_failure()

from cognitia.bootstrap.stack import CognitiaStack

stack = CognitiaStack.create(
    prompts_dir=Path("prompts"),
    skills_dir=Path("skills"),
    project_root=Path("."),
    runtime_config=RuntimeConfig(...),
    sandbox_provider=...,           # SandboxProvider | None
    web_provider=...,               # WebProvider | None
    todo_provider=...,              # TodoProvider | None
    memory_bank_provider=...,       # MemoryBankProvider | None
    plan_manager=...,               # PlanManager | None
    thinking_enabled=True,          # bool
    allowed_system_tools={...},     # set[str]
    tool_budget_config=...,         # ToolBudgetConfig | None
    escalate_roles={...},           # set[str]
    local_tool_resolver=...,        # object | None
)

from cognitia.policy import DefaultToolPolicy

policy = DefaultToolPolicy(
    allowed_system_tools={"bash", "read"},
    extra_denied={"dangerous_tool"},
)

result = policy.can_use_tool(tool_name, input_data, state)

from cognitia.policy import DefaultToolIdCodec

codec = DefaultToolIdCodec()
codec.encode("server", "tool")      # "mcp__server__tool"
codec.extract_server("mcp__s__t")   # "s"
codec.matches("mcp__s__t", "s")     # True

from cognitia.routing import KeywordRoleRouter

router = KeywordRoleRouter(
    default_role="coach",
    keyword_map={
        "deposit_advisor": ["deposit", "savings"],
        "credit_advisor": ["credit", "loan"],
    },
)

router.resolve("I want a savings account")  # "deposit_advisor"
router.resolve("Hello")                      # "coach"
router.resolve("...", explicit_role="coach")  # "coach" (explicit wins)

from cognitia.commands import CommandRegistry

registry = CommandRegistry()
registry.register(name="help", handler=handler, aliases=["h", "?"], description="Show help")
result = await registry.dispatch("/help", context)