VisvoAI Docs

API reference

Every public signature in visvoai-ai and visvoai-core

Grouped by module, matching the source tree. Everything here is exported from visvoai.ai.__init__ or visvoai.core.__init__ unless a full submodule path is shown.

visvoai-ai

visvoai.ai.resolve

def build_chat_model(
    deployment_id: str,
    *,
    level: Optional[str] = None,
    thinking_raw: Optional[dict] = None,
    api_key: Optional[str] = None,
    base_url: Optional[str] = None,
    codec: IdentityCodec = DEFAULT_CODEC,
) -> BaseChatModel: ...

def cost_of(
    deployment_id: str, input_tokens: int, output_tokens: int,
    codec: IdentityCodec = DEFAULT_CODEC,
) -> float: ...

visvoai.ai.usage

def usage_from(message_or_chunk: Any) -> dict:
    """{'input': int, 'output': int, 'total': int} — 0 when usage_metadata absent."""

visvoai.ai.search

def run_search(
    query: str, *, deployment_id: Optional[str] = None,
    provider: Optional[str] = None, system: Optional[str] = None,
    api_key: Optional[str] = None, codec: IdentityCodec = DEFAULT_CODEC,
) -> SearchResult: ...

def fetch_url(
    url: str, *, deployment_id: Optional[str] = None,
    provider: Optional[str] = None, api_key: Optional[str] = None,
    codec: IdentityCodec = DEFAULT_CODEC,
) -> str: ...

class FetchError(Exception): ...

@dataclass
class SearchSource:
    title: str
    url: str
    snippet: str = ""

@dataclass
class SearchResult:
    text: str
    sources: List[SearchSource] = field(default_factory=list)
    queries: List[str] = field(default_factory=list)

visvoai.ai.thinking

class ThinkingLevel(str, Enum):
    OFF = "off"; LOW = "low"; MEDIUM = "medium"; HIGH = "high"

def resolve_level(value: Any, default: ThinkingLevel = ThinkingLevel.OFF) -> ThinkingLevel: ...

def thinking_kwargs(mechanism: ThinkingMechanism, level: ThinkingLevel) -> Dict[str, Any]: ...

visvoai.ai.identity

class DeploymentId(NamedTuple):
    provider: str
    model: str
    effort: Optional[str] = None

class IdentityCodec(Protocol):
    def build(self, provider: str, model: str, effort: Optional[str] = None) -> str: ...
    def parse(self, deployment_id: str) -> DeploymentId: ...

class ColonAtCodec:
    """provider:model[@effort] — the default codec (DEFAULT_CODEC)."""

visvoai.ai.deployments

@dataclass(frozen=True)
class DeploymentInfo:
    id: str
    model: str
    display_name: str
    provider: str
    family: str
    capabilities: List[Capability]
    reasoning: bool
    input_cost_per_million: float
    output_cost_per_million: float
    context_window: int
    supports_thinking: bool
    thinking_levels: List[ThinkingLevel]
    default_thinking: ThinkingLevel

class DeploymentRegistry:
    def __init__(self, model_defs: List[ModelDefinition]) -> None: ...
    def get_model(self, model_id: str) -> Optional[Model]: ...
    def get_deployment(self, deployment_id: str, codec=DEFAULT_CODEC) -> Optional[Deployment]: ...
    def deployments_for(self, model_id: str) -> List[Deployment]: ...
    def list_deployments(self, capability=None, codec=DEFAULT_CODEC) -> List[DeploymentInfo]: ...
    def get_deployment_info(self, deployment_id: str, codec=DEFAULT_CODEC) -> Optional[DeploymentInfo]: ...
    def default_deployment(self, capability=Capability.CHAT, codec=DEFAULT_CODEC,
                           provider: Optional[str] = None) -> Optional[str]: ...

# module-level free functions over a default DeploymentRegistry instance:
def get_deployment(deployment_id: str, codec=DEFAULT_CODEC) -> Optional[Deployment]: ...
def deployments_for(model_id: str) -> List[Deployment]: ...
def list_deployments(capability=None, codec=DEFAULT_CODEC) -> List[DeploymentInfo]: ...
def get_deployment_info(deployment_id: str, codec=DEFAULT_CODEC) -> Optional[DeploymentInfo]: ...
def default_deployment(capability=Capability.CHAT, codec=DEFAULT_CODEC,
                       provider: Optional[str] = None) -> Optional[str]: ...
def install_catalog(model_defs: List[ModelDefinition]) -> None: ...
def set_default_registry(reg: DeploymentRegistry) -> None: ...
def get_default_registry() -> DeploymentRegistry: ...

visvoai.ai.model_registry

# NOTE: this is the `get_model` exported as `visvoai.ai.get_model` — it looks up
# raw registry data by api_id. (deployments.py has a same-named function
# returning Model, but it is deliberately not exported.)
def get_model(api_id: str) -> Optional[ModelDefinition]: ...

class Capability(str, Enum):
    CHAT = "chat"; SEARCH = "search"
    IMAGE_GEN = "image_gen"; AUDIO_GEN = "audio_gen"; EMBEDDING = "embedding"

@dataclass
class ModelDefinition:
    api_id: str
    display_name: str
    input_cost_per_million: float
    output_cost_per_million: float
    context_window: int = 0
    cache_read_cost_per_million: float = 0.0
    search_query_cost: float = 0.0
    search_billed_per_request: bool = False
    provider: str = "gemini"
    icon_url: str = "https://www.google.com/favicon.ico"
    supports_thinking: bool = False
    enabled: bool = True
    deprecated: bool = False
    default: bool = False
    default_thinking_label: Optional[str] = None
    capabilities: List[Capability] = field(default_factory=lambda: [Capability.CHAT])
    unit_cost: Optional[float] = None
    unit: Optional[str] = None
    base_url: Optional[str] = None
    key_env: Optional[str] = None

MODELS: List[ModelDefinition]              # the baked rate card
DEFAULT_MODEL_FOR: Dict[Capability, str]   # e.g. {SEARCH: "gemini-3-flash-preview"}

visvoai.ai.providers.base

class NotSupported(NotImplementedError): ...

def default_content_events(chunk: Any) -> Generator[Dict[str, Any], None, None]: ...

class Provider(ABC):
    def build(self, slug: str, api_key=None, base_url=None, **extra) -> BaseChatModel: ...
    def normalize_content(self, chunk: Any) -> Generator[Dict[str, Any], None, None]: ...
    def search(self, query: str, *, slug: str, api_key=None, system=None) -> "SearchResult": ...
    def fetch_url(self, url: str, *, slug: str, api_key=None) -> str: ...

visvoai.ai.providers.factory / .config

def get_provider(provider_name: Optional[str]) -> Provider: ...
def get_provider_for_model(model_id: str, capability: Optional[Capability] = None) -> Provider: ...

def resolve_api_key(provider: str, api_key: Optional[str] = None,
                    env_var: Optional[str] = None) -> str: ...
def resolve_base_url(provider: str, base_url: Optional[str] = None) -> Optional[str]: ...

visvoai.ai.catalog

class CatalogSource(ABC):
    def models(self) -> List[ModelDefinition]: ...

class BakedSource(CatalogSource):
    def __init__(self, models: Optional[List[ModelDefinition]] = None) -> None: ...

Gate = Callable[[ModelDefinition], bool]

def build_catalog(sources: List[CatalogSource], gate: Optional[Gate] = None) -> List[ModelDefinition]: ...
def validate(defs: List[ModelDefinition]) -> None: ...

class ModelsDevSource(CatalogSource):
    def __init__(self, catalog: Dict[str, Any]) -> None: ...

def to_definitions(catalog: Dict[str, Any]) -> List[ModelDefinition]: ...

class RemoteModelsDevSource(CatalogSource):
    def __init__(self, cache_path, *, url=DEFAULT_URL, ttl_seconds=DEFAULT_TTL_SECONDS,
                timeout=20, fetcher=None, use_bundled_fallback=True) -> None: ...

visvoai-core

visvoai.core.runtime

class AgentRuntime:
    def build_graph(
        self, model: BaseChatModel, core_tools: List[Any],
        all_tools_map: Optional[Dict[str, Any]] = None,
        system_prompt: str = "",
        checkpointer: Optional[BaseCheckpointSaver] = None,
        tool_configs: Optional[Dict[str, Any]] = None,
        lean_prompt: bool = False,
        per_round_retrieve: Optional[Any] = None,
    ): ...

    def _extend_graph(self, workflow: StateGraph, tool_configs: dict) -> None: ...
    def _build_agent_node(self, ctx: GraphBuildContext): ...
    def _build_tools_node(self, ctx: GraphBuildContext): ...
    def _agent_routing(self, ctx: GraphBuildContext): ...
    def _tools_routing(self, tool_configs: dict): ...
    def _get_checkpointer(self, checkpointer=None) -> Optional[BaseCheckpointSaver]: ...
    def _get_state_class(self) -> type: ...
    def _get_interrupt_nodes(self) -> Optional[List[str]]: ...

visvoai.core.graph

DEFAULT_MAX_AGENT_STEPS = 10

@dataclass(frozen=True)
class GraphBuildContext:
    model: BaseChatModel
    core_tools: List[BaseTool]
    all_tools_map: Dict[str, BaseTool]
    all_tools: List[BaseTool]
    system_prompt: str
    tool_configs: Dict[str, Any] = field(default_factory=dict)
    per_round_retrieve: Optional[Any] = None
    lean_prompt: bool = False
    max_agent_steps: Optional[int] = None

def build_graph(
    model, core_tools, all_tools_map=None, system_prompt="",
    checkpointer=None, tool_configs=None, lean_prompt=False,
    per_round_retrieve=None, max_agent_steps=DEFAULT_MAX_AGENT_STEPS,
    _runtime=None,
): ...

visvoai.core.state

class AgentState(TypedDict, total=False):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    active_mcp_tools: Annotated[List[str], _union_ordered]

visvoai.core.context

@dataclass
class RuntimeContext:
    request_id: Optional[str] = None
    subagent_depth: int = 0
    parent_tool_call_id: Optional[str] = None

visvoai.core.tools

class ToolConfig:
    is_core: bool = False
    no_cache: bool = False
    cache_key_args: Optional[List[str]] = None
    routing_hint: Optional[str] = None
    anti_patterns: Optional[List[str]] = None
    depends_on: Optional[List[str]] = None
    parallel_with: Optional[List[str]] = None
    skip_context_chunk: bool = False
    persist_context: bool = False
    sequential_only: bool = False
    idempotent: bool = True
    deprecated: bool = False
    disabled: bool = False

def tool_config(**kwargs: Any): ...   # class decorator

class BaseAgentTool(ToolConfig, ABC):
    name: str
    description: str
    args_schema: Type[BaseModel]
    llm_schema: Optional[Type[BaseModel]] = None
    _persistence: ToolPersistence = ToolPersistence()
    _registry: ClassVar[List[Type["BaseAgentTool"]]] = []

    def _execute(self, tool_call_id: str, **kwargs: Any) -> Any: ...  # abstract
    def execute(self, tool_call_id=None, agent_step=0,
               execution_phase=None, **kwargs: Any) -> Any: ...

visvoai.core.adapt

ToolLike = Union[BaseTool, BaseAgentTool, Type[BaseAgentTool], Callable[..., Any]]

def as_tool(obj: ToolLike) -> BaseTool: ...
def as_tools(objs: Iterable[ToolLike]) -> List[BaseTool]: ...
def as_tools_map(objs: Iterable[ToolLike]) -> Dict[str, BaseTool]: ...
async def ask(graph: Any, text: str, thread_id: Optional[str] = None) -> str: ...

visvoai.core.results

class ToolStatus(str, Enum):
    SUCCESS = "SUCCESS"; EMPTY_RESULT = "EMPTY_RESULT"
    INVALID_INPUT = "INVALID_INPUT"; TOOL_ERROR = "TOOL_ERROR"

class ToolResult(BaseModel):
    tool_name: str
    status: ToolStatus
    result: str
    data: Optional[Dict[str, Any]] = None

    @classmethod
    def success(cls, tool_name, payload, display=None, **meta) -> "ToolResult": ...
    @classmethod
    def invalid_input(cls, tool_name, whats_wrong, expected=None) -> "ToolResult": ...
    @classmethod
    def tool_error(cls, tool_name, error, **meta) -> "ToolResult": ...
    @classmethod
    def empty(cls, tool_name, reason, next_step=None, **meta) -> "ToolResult": ...

visvoai.core.persistence

class ToolPersistence:
    def on_start(self, *, tool_id, message_id, tool_name, tool_input,
                 agent_step, execution_phase=None, **kwargs) -> str: ...
    def on_resume(self, tool_id: str) -> str: ...
    def on_complete(self, *, tool_id, status, output, duration_ms, **kwargs) -> None: ...
    def on_error(self, *, tool_id, error, duration_ms) -> None: ...

class LLMPersistence:
    def on_call_complete(self, *, message_id, model_name, input_tokens,
                         output_tokens, total_tokens, action,
                         estimated_cost_usd, tool_call_id=None) -> None: ...
    def on_thinking_log(self, *, message_id, thinking_text,
                        tool_call_id=None) -> None: ...

visvoai.core.retrieval

class ToolCatalog:
    K1 = 1.5
    B = 0.75
    def __init__(self, entries: List[Tuple]) -> None: ...
    def search(self, query: str, k: int = 8,
              query_vec: Optional[List[float]] = None) -> List[str]: ...

def build_catalog_from_servers(servers: List) -> ToolCatalog: ...

def make_per_round_retrieve(
    catalog: ToolCatalog, k: int = 8,
    embed_query: Optional[Callable[[str], Optional[List[float]]]] = None,
) -> Callable[[str], List[str]]: ...

See the individual guide pages for narrative explanations of every signature above: Providers & the model registry, Cost, usage & thinking levels, The agent loop, Defining tools, Extension seams, Persistence, Tool retrieval at scale.

On this page