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] = Nonevisvoai.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.