925 lines
34 KiB
Python
925 lines
34 KiB
Python
"""Sync Honcho client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from collections.abc import Generator, Mapping, Sequence
|
|
from typing import Any, Literal
|
|
|
|
import httpx
|
|
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, validate_call
|
|
|
|
from .aio import HonchoAio
|
|
from .api_types import (
|
|
MessageResponse,
|
|
PeerConfig,
|
|
PeerResponse,
|
|
QueueStatusResponse,
|
|
ScopeResponse,
|
|
SessionConfiguration,
|
|
SessionPeerConfig,
|
|
SessionResponse,
|
|
WorkspaceConfiguration,
|
|
WorkspaceResponse,
|
|
)
|
|
from .base import PeerBase, ScopeBase, SessionBase
|
|
from .http import AsyncHonchoHTTPClient, HonchoHTTPClient, routes
|
|
from .message import Message
|
|
from .mixins import MetadataConfigMixin
|
|
from .pagination import SyncPage
|
|
from .peer import Peer, serialize_response_format
|
|
from .scope import Scope
|
|
from .session import Session
|
|
from .types import DialecticStreamResponse
|
|
from .utils import (
|
|
normalize_peers_to_dict,
|
|
parse_sse_stream,
|
|
resolve_id,
|
|
validate_scope_id,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Environment configuration
|
|
ENVIRONMENTS = {
|
|
"local": "http://localhost:8000",
|
|
"production": "https://api.honcho.dev",
|
|
}
|
|
|
|
|
|
class Honcho(BaseModel, MetadataConfigMixin): # pyright: ignore[reportUnsafeMultipleInheritance]
|
|
"""
|
|
Main client for the Honcho SDK.
|
|
|
|
Provides access to peers, sessions, and workspace operations with configuration
|
|
from environment variables or explicit parameters. This is the primary entry
|
|
point for interacting with the Honcho conversational memory platform.
|
|
|
|
Attributes:
|
|
workspace_id: Workspace ID for scoping operations
|
|
metadata: Cached metadata for this workspace. May be stale if not recently
|
|
fetched. Call get_metadata() for fresh data.
|
|
configuration: Cached configuration for this workspace. May be stale if not
|
|
recently fetched. Call get_configuration() for fresh data.
|
|
"""
|
|
|
|
model_config = ConfigDict(extra="forbid") # pyright: ignore
|
|
|
|
workspace_id: str = Field(
|
|
...,
|
|
min_length=1,
|
|
description="Workspace ID for scoping operations",
|
|
)
|
|
_metadata: dict[str, object] | None = PrivateAttr(default=None)
|
|
_configuration: WorkspaceConfiguration | None = PrivateAttr(default=None)
|
|
_http: HonchoHTTPClient = PrivateAttr()
|
|
_async_http: AsyncHonchoHTTPClient | None = PrivateAttr(default=None)
|
|
_http_config: dict[str, Any] = PrivateAttr()
|
|
_base_url: str = PrivateAttr()
|
|
_workspace_ensured: bool = PrivateAttr(default=False)
|
|
|
|
@property
|
|
def metadata(self) -> dict[str, object] | None:
|
|
"""Cached metadata for this workspace. May be stale. Use get_metadata() for fresh data."""
|
|
return self._metadata
|
|
|
|
@property
|
|
def configuration(self) -> WorkspaceConfiguration | None:
|
|
"""Cached configuration for this workspace. May be stale. Use get_configuration() for fresh data."""
|
|
return self._configuration
|
|
|
|
# MetadataConfigMixin implementation
|
|
def _get_http_client(self):
|
|
"""Return the sync HTTP client used by metadata helpers."""
|
|
return self._http
|
|
|
|
def _get_fetch_route(self) -> str:
|
|
"""Return the workspace fetch route for metadata helpers."""
|
|
return routes.workspaces()
|
|
|
|
def _get_update_route(self) -> str:
|
|
"""Return the workspace update route for metadata helpers."""
|
|
return routes.workspace(self.workspace_id)
|
|
|
|
def _get_fetch_body(self) -> dict[str, Any]:
|
|
"""Return the request body used to fetch this workspace."""
|
|
return {"id": self.workspace_id}
|
|
|
|
def _parse_response(
|
|
self, data: dict[str, Any]
|
|
) -> tuple[dict[str, object], dict[str, object]]:
|
|
"""Parse workspace metadata and configuration from an API response."""
|
|
workspace = WorkspaceResponse.model_validate(data)
|
|
# Return configuration as dict for mixin compatibility
|
|
return workspace.metadata or {}, workspace.configuration.model_dump(
|
|
exclude_none=True
|
|
)
|
|
|
|
def get_configuration(self) -> WorkspaceConfiguration: # pyright: ignore[reportIncompatibleMethodOverride]
|
|
"""
|
|
Get configuration from the server and update the cache.
|
|
|
|
Returns:
|
|
A WorkspaceConfiguration object containing the configuration settings.
|
|
"""
|
|
data = self._get_http_client().post(
|
|
self._get_fetch_route(), body=self._get_fetch_body()
|
|
)
|
|
workspace = WorkspaceResponse.model_validate(data)
|
|
self._metadata = workspace.metadata or {}
|
|
self._configuration = workspace.configuration
|
|
return self._configuration
|
|
|
|
@validate_call
|
|
def set_configuration( # pyright: ignore[reportIncompatibleMethodOverride]
|
|
self,
|
|
configuration: WorkspaceConfiguration = Field(
|
|
..., description="Configuration to set"
|
|
),
|
|
) -> None:
|
|
"""
|
|
Set configuration on the server and update the cache.
|
|
|
|
Args:
|
|
configuration: A WorkspaceConfiguration object with configuration settings.
|
|
"""
|
|
self._get_http_client().put(
|
|
self._get_update_route(),
|
|
body={"configuration": configuration.model_dump(exclude_none=True)},
|
|
)
|
|
self._configuration = configuration # pyright: ignore[reportIncompatibleVariableOverride]
|
|
|
|
@property
|
|
def base_url(self) -> str:
|
|
"""The base URL of the Honcho API."""
|
|
return self._base_url
|
|
|
|
@property
|
|
def _async_http_client(self) -> AsyncHonchoHTTPClient:
|
|
"""Lazily create and return the async HTTP client."""
|
|
if self._async_http is None:
|
|
self._async_http = AsyncHonchoHTTPClient(**self._http_config)
|
|
return self._async_http
|
|
|
|
@property
|
|
def aio(self) -> HonchoAio:
|
|
"""
|
|
Access async versions of all Honcho methods.
|
|
|
|
Returns an HonchoAio view that provides async versions of all methods
|
|
while sharing state with this Honcho instance.
|
|
|
|
Example:
|
|
```python
|
|
honcho = Honcho(workspace_id="my-workspace")
|
|
|
|
# Async operations
|
|
peer = await honcho.aio.peer("user-123")
|
|
async for p in honcho.aio.peers():
|
|
print(p.id)
|
|
```
|
|
"""
|
|
return HonchoAio(self)
|
|
|
|
@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
|
|
def __init__(
|
|
self,
|
|
api_key: str | None = None,
|
|
environment: Literal["local", "production"] | None = None,
|
|
base_url: str | None = Field(None, description="Base URL for the Honcho API"),
|
|
workspace_id: str | None = Field(
|
|
None, min_length=1, description="Workspace ID for scoping operations"
|
|
),
|
|
timeout: float | None = Field(None, gt=0, description="Timeout in seconds"),
|
|
max_retries: int | None = Field(
|
|
None, ge=0, description="Maximum number of retries"
|
|
),
|
|
default_headers: Mapping[str, str] | None = None,
|
|
default_query: Mapping[str, object] | None = None,
|
|
http_client: httpx.Client | None = Field(
|
|
None, description="Custom HTTP client"
|
|
),
|
|
) -> None:
|
|
"""
|
|
Initialize the Honcho client.
|
|
|
|
Args:
|
|
api_key:
|
|
API key for authentication. If not provided, will attempt to
|
|
read from HONCHO_API_KEY environment variable
|
|
environment:
|
|
Environment to use (local or production)
|
|
base_url:
|
|
Base URL for the Honcho API. If not provided, will attempt to
|
|
read from HONCHO_URL environment variable or default to the
|
|
production API URL
|
|
workspace_id:
|
|
Workspace ID to use for operations. If not provided, will
|
|
attempt to read from HONCHO_WORKSPACE_ID environment variable
|
|
or default to "default"
|
|
timeout:
|
|
Optional custom timeout for the HTTP client.
|
|
max_retries:
|
|
Optional custom maximum number of retries for the HTTP client.
|
|
default_headers:
|
|
Optional custom default headers for the HTTP client.
|
|
default_query:
|
|
Optional custom default query parameters for the HTTP client.
|
|
http_client:
|
|
Optional custom httpx client.
|
|
"""
|
|
# Resolve workspace_id before calling super().__init__
|
|
resolved_workspace_id = workspace_id or os.getenv(
|
|
"HONCHO_WORKSPACE_ID", "default"
|
|
)
|
|
|
|
super().__init__(workspace_id=resolved_workspace_id)
|
|
|
|
# Resolve API key
|
|
resolved_api_key = api_key or os.getenv("HONCHO_API_KEY")
|
|
|
|
# Resolve base URL
|
|
if base_url:
|
|
resolved_base_url = base_url
|
|
elif environment:
|
|
resolved_base_url = ENVIRONMENTS[environment]
|
|
else:
|
|
resolved_base_url = os.getenv("HONCHO_URL", ENVIRONMENTS["production"])
|
|
|
|
self._base_url = resolved_base_url
|
|
|
|
# Build HTTP client kwargs
|
|
http_kwargs: dict[str, Any] = {
|
|
"base_url": resolved_base_url,
|
|
"api_key": resolved_api_key,
|
|
}
|
|
|
|
if timeout is not None:
|
|
http_kwargs["timeout"] = timeout
|
|
if max_retries is not None:
|
|
http_kwargs["max_retries"] = max_retries
|
|
if default_headers is not None:
|
|
http_kwargs["default_headers"] = dict(default_headers)
|
|
if default_query is not None:
|
|
http_kwargs["default_query"] = dict(default_query)
|
|
|
|
# Store config for lazy async client creation (without custom http_client)
|
|
self._http_config = dict(http_kwargs)
|
|
|
|
if http_client is not None:
|
|
http_kwargs["http_client"] = http_client
|
|
|
|
self._http = HonchoHTTPClient(**http_kwargs)
|
|
|
|
def _ensure_workspace(self) -> None:
|
|
"""
|
|
Ensure the workspace exists on the server.
|
|
|
|
The Honcho API uses get-or-create semantics for workspaces via
|
|
`POST /v3/workspaces`. This SDK uses that endpoint once per client
|
|
instance to guarantee that subsequent workspace-scoped calls (peers,
|
|
sessions, queue status, etc.) operate on an existing workspace.
|
|
"""
|
|
if self._workspace_ensured:
|
|
return
|
|
self._http.post(routes.workspaces(), body={"id": self.workspace_id})
|
|
self._workspace_ensured = True
|
|
|
|
async def _ensure_workspace_async(self) -> None:
|
|
"""
|
|
Async version of `_ensure_workspace`.
|
|
|
|
This performs the same get-or-create call, but via the async HTTP client
|
|
used by `honcho.aio`.
|
|
"""
|
|
if self._workspace_ensured:
|
|
return
|
|
await self._async_http_client.post(
|
|
routes.workspaces(), body={"id": self.workspace_id}
|
|
)
|
|
self._workspace_ensured = True
|
|
|
|
@validate_call
|
|
def peer(
|
|
self,
|
|
id: str = Field(
|
|
..., min_length=1, description="Unique identifier for the peer"
|
|
),
|
|
*,
|
|
metadata: dict[str, object] | None = Field(
|
|
None,
|
|
description="Optional metadata dictionary to associate with this peer. If set, will get/create peer immediately with metadata.",
|
|
),
|
|
configuration: PeerConfig | None = Field(
|
|
None,
|
|
description="Optional configuration to set for this peer. If set, will get/create peer immediately with flags.",
|
|
),
|
|
) -> Peer:
|
|
"""
|
|
Get or create a peer with the given ID.
|
|
|
|
Makes an API call to get or create the peer, then returns a Peer object
|
|
with cached values from the response.
|
|
|
|
Args:
|
|
id: Unique identifier for the peer within the workspace.
|
|
metadata: Optional metadata dictionary to associate with this peer.
|
|
configuration: Optional configuration to set for this peer.
|
|
|
|
Returns:
|
|
A Peer object with cached metadata, configuration, and created_at.
|
|
"""
|
|
self._ensure_workspace()
|
|
body: dict[str, Any] = {"id": id}
|
|
if metadata is not None:
|
|
body["metadata"] = metadata
|
|
if configuration is not None:
|
|
body["configuration"] = configuration.model_dump(exclude_none=True)
|
|
|
|
data = self._http.post(routes.peers(self.workspace_id), body=body)
|
|
peer_data = PeerResponse.model_validate(data)
|
|
return Peer(
|
|
id,
|
|
self,
|
|
metadata=peer_data.metadata,
|
|
configuration=peer_data.configuration,
|
|
created_at=peer_data.created_at,
|
|
)
|
|
|
|
def peers(
|
|
self,
|
|
filters: dict[str, object] | None = None,
|
|
*,
|
|
page: int = 1,
|
|
size: int = 50,
|
|
reverse: bool = False,
|
|
) -> SyncPage[PeerResponse, Peer]:
|
|
"""
|
|
Get all peers in the current workspace.
|
|
|
|
Args:
|
|
filters: Optional filter criteria.
|
|
page: Page number (1-indexed). Default: 1.
|
|
size: Number of items per page. Default: 50.
|
|
reverse: If True, reverses the default ordering. Default: False.
|
|
|
|
Returns:
|
|
A SyncPage of Peer objects representing all peers in the workspace
|
|
"""
|
|
self._ensure_workspace()
|
|
query: dict[str, Any] = {"page": page, "size": size}
|
|
if reverse:
|
|
query["reverse"] = "true"
|
|
data = self._http.post(
|
|
routes.peers_list(self.workspace_id),
|
|
body={"filters": filters} if filters else None,
|
|
query=query,
|
|
)
|
|
|
|
def transform(peer: PeerResponse) -> Peer:
|
|
"""Convert a peer API response into a Peer SDK object."""
|
|
return Peer(
|
|
peer.id,
|
|
self,
|
|
metadata=peer.metadata,
|
|
configuration=peer.configuration,
|
|
created_at=peer.created_at,
|
|
)
|
|
|
|
def fetch_next(next_page: int) -> SyncPage[PeerResponse, Peer]:
|
|
"""Fetch the next page while preserving filters and ordering."""
|
|
next_query: dict[str, Any] = {"page": next_page, "size": size}
|
|
if reverse:
|
|
next_query["reverse"] = "true"
|
|
next_data = self._http.post(
|
|
routes.peers_list(self.workspace_id),
|
|
body={"filters": filters} if filters else None,
|
|
query=next_query,
|
|
)
|
|
return SyncPage(next_data, PeerResponse, transform, fetch_next)
|
|
|
|
return SyncPage(data, PeerResponse, transform, fetch_next)
|
|
|
|
@validate_call
|
|
def session(
|
|
self,
|
|
id: str = Field(
|
|
..., min_length=1, description="Unique identifier for the session"
|
|
),
|
|
*,
|
|
metadata: dict[str, object] | None = Field(
|
|
None,
|
|
description="Optional metadata dictionary to associate with this session. If set, will get/create session immediately with metadata.",
|
|
),
|
|
configuration: SessionConfiguration | None = Field(
|
|
None,
|
|
description="Optional configuration to set for this session. If set, will get/create session immediately with flags.",
|
|
),
|
|
peers: str
|
|
| PeerBase
|
|
| tuple[str, SessionPeerConfig]
|
|
| tuple[PeerBase, SessionPeerConfig]
|
|
| list[PeerBase | str]
|
|
| list[tuple[PeerBase | str, SessionPeerConfig]]
|
|
| list[PeerBase | str | tuple[PeerBase | str, SessionPeerConfig]]
|
|
| None = Field(
|
|
None,
|
|
description="Optional peers to attach to the session at creation. Accepts the same shape as Session.add_peers.",
|
|
),
|
|
scopes: Sequence[str | ScopeBase] | None = Field(
|
|
None,
|
|
description="Optional scopes this session should join. Each scope is created if it does not exist yet.",
|
|
),
|
|
) -> Session:
|
|
"""
|
|
Get or create a session with the given ID.
|
|
|
|
Makes an API call to get or create the session, then returns a Session object
|
|
with cached values from the response.
|
|
|
|
Args:
|
|
id: Unique identifier for the session within the workspace.
|
|
metadata: Optional metadata dictionary to associate with this session.
|
|
configuration: Optional configuration to set for this session.
|
|
peers: Optional peers to attach to the session at creation. Accepts the
|
|
same shape as Session.add_peers (peer ID string, Peer object, list
|
|
of either, or tuples with SessionPeerConfig).
|
|
scopes: Optional scopes this session should join, as IDs or Scope
|
|
objects. Each scope is created if it does not exist yet. Attaching
|
|
at creation avoids the asynchronous backfill a later
|
|
``scope.add_sessions()`` triggers, since there is no history to
|
|
copy.
|
|
|
|
Returns:
|
|
A Session object with cached metadata, configuration, created_at, and is_active.
|
|
"""
|
|
self._ensure_workspace()
|
|
body: dict[str, Any] = {"id": id}
|
|
if metadata is not None:
|
|
body["metadata"] = metadata
|
|
if configuration is not None:
|
|
body["configuration"] = configuration.model_dump(exclude_none=True)
|
|
if peers is not None:
|
|
body["peers"] = normalize_peers_to_dict(peers)
|
|
if scopes is not None:
|
|
body["scopes"] = [validate_scope_id(resolve_id(scope)) for scope in scopes]
|
|
|
|
data = self._http.post(routes.sessions(self.workspace_id), body=body)
|
|
session_data = SessionResponse.model_validate(data)
|
|
return Session(
|
|
id,
|
|
self,
|
|
metadata=session_data.metadata,
|
|
configuration=SessionConfiguration.model_validate(
|
|
session_data.configuration.model_dump()
|
|
),
|
|
created_at=session_data.created_at,
|
|
is_active=session_data.is_active,
|
|
)
|
|
|
|
def sessions(
|
|
self,
|
|
filters: dict[str, object] | None = None,
|
|
*,
|
|
page: int = 1,
|
|
size: int = 50,
|
|
reverse: bool = False,
|
|
) -> SyncPage[SessionResponse, Session]:
|
|
"""
|
|
Get all sessions in the current workspace.
|
|
|
|
Args:
|
|
filters: Optional filter criteria.
|
|
page: Page number (1-indexed). Default: 1.
|
|
size: Number of items per page. Default: 50.
|
|
reverse: If True, reverses the default ordering. Default: False.
|
|
|
|
Returns:
|
|
A SyncPage of Session objects representing all sessions in the workspace.
|
|
"""
|
|
self._ensure_workspace()
|
|
query: dict[str, Any] = {"page": page, "size": size}
|
|
if reverse:
|
|
query["reverse"] = "true"
|
|
data = self._http.post(
|
|
routes.sessions_list(self.workspace_id),
|
|
body={"filters": filters} if filters else None,
|
|
query=query,
|
|
)
|
|
|
|
def transform(session: SessionResponse) -> Session:
|
|
"""Convert a session API response into a Session SDK object."""
|
|
return Session(
|
|
session.id,
|
|
self,
|
|
metadata=session.metadata,
|
|
configuration=session.configuration,
|
|
created_at=session.created_at,
|
|
is_active=session.is_active,
|
|
)
|
|
|
|
def fetch_next(next_page: int) -> SyncPage[SessionResponse, Session]:
|
|
"""Fetch the next page while preserving filters and ordering."""
|
|
next_query: dict[str, Any] = {"page": next_page, "size": size}
|
|
if reverse:
|
|
next_query["reverse"] = "true"
|
|
next_data = self._http.post(
|
|
routes.sessions_list(self.workspace_id),
|
|
body={"filters": filters} if filters else None,
|
|
query=next_query,
|
|
)
|
|
return SyncPage(next_data, SessionResponse, transform, fetch_next)
|
|
|
|
return SyncPage(data, SessionResponse, transform, fetch_next)
|
|
|
|
@validate_call
|
|
def scope(
|
|
self,
|
|
id: str = Field( # noqa: A002
|
|
..., min_length=1, description="Unprefixed name for the scope"
|
|
),
|
|
*,
|
|
metadata: dict[str, object] | None = Field(
|
|
None,
|
|
description="Optional metadata dictionary to associate with this scope.",
|
|
),
|
|
) -> Scope:
|
|
"""
|
|
Get or create a scope with the given ID.
|
|
|
|
A scope is a named set of sessions that acts as a visibility boundary:
|
|
recall performed through the scope sees only what happened in its sessions,
|
|
while the underlying peer keeps its single unified representation of
|
|
everything.
|
|
|
|
Args:
|
|
id: Unprefixed scope name, unique within the workspace.
|
|
metadata: Optional metadata dictionary to associate with this scope.
|
|
|
|
Returns:
|
|
A Scope object for managing membership.
|
|
|
|
Raises:
|
|
ValueError: If the scope ID is invalid.
|
|
|
|
Example:
|
|
```python
|
|
therapy = honcho.scope("therapy")
|
|
therapy.add_sessions([session_1, session_2])
|
|
```
|
|
"""
|
|
validate_scope_id(id)
|
|
self._ensure_workspace()
|
|
body: dict[str, Any] = {"id": id}
|
|
if metadata is not None:
|
|
body["metadata"] = metadata
|
|
|
|
data = self._http.post(routes.scopes(self.workspace_id), body=body)
|
|
scope_data = ScopeResponse.model_validate(data)
|
|
return Scope(
|
|
id,
|
|
self,
|
|
metadata=scope_data.metadata,
|
|
created_at=scope_data.created_at,
|
|
)
|
|
|
|
def scopes(
|
|
self,
|
|
*,
|
|
page: int = 1,
|
|
size: int = 50,
|
|
reverse: bool = False,
|
|
) -> SyncPage[ScopeResponse, Scope]:
|
|
"""
|
|
Get all scopes in the current workspace.
|
|
|
|
Args:
|
|
page: Page number (1-indexed). Default: 1.
|
|
size: Number of items per page. Default: 50.
|
|
reverse: If True, reverses the default ordering. Default: False.
|
|
|
|
Returns:
|
|
A SyncPage of Scope objects representing all scopes in the workspace.
|
|
"""
|
|
self._ensure_workspace()
|
|
|
|
def fetch(next_page: int) -> dict[str, Any]:
|
|
query: dict[str, Any] = {"page": next_page, "size": size}
|
|
if reverse:
|
|
query["reverse"] = "true"
|
|
return self._http.post(routes.scopes_list(self.workspace_id), query=query)
|
|
|
|
def transform(scope: ScopeResponse) -> Scope:
|
|
"""Convert a scope API response into a Scope SDK object."""
|
|
return Scope(
|
|
scope.id,
|
|
self,
|
|
metadata=scope.metadata,
|
|
created_at=scope.created_at,
|
|
)
|
|
|
|
def fetch_next(next_page: int) -> SyncPage[ScopeResponse, Scope]:
|
|
return SyncPage(fetch(next_page), ScopeResponse, transform, fetch_next)
|
|
|
|
return SyncPage(fetch(page), ScopeResponse, transform, fetch_next)
|
|
|
|
def workspaces(
|
|
self,
|
|
filters: dict[str, object] | None = None,
|
|
*,
|
|
page: int = 1,
|
|
size: int = 50,
|
|
reverse: bool = False,
|
|
) -> SyncPage[WorkspaceResponse, str]:
|
|
"""
|
|
Get all workspace IDs from the Honcho instance.
|
|
|
|
Makes an API call to retrieve all workspace IDs that the authenticated
|
|
user has access to.
|
|
|
|
Args:
|
|
filters: Optional filter criteria.
|
|
page: Page number (1-indexed). Default: 1.
|
|
size: Number of items per page. Default: 50.
|
|
reverse: If True, reverses the default ordering. Default: False.
|
|
|
|
Returns:
|
|
A paginated SyncPage of workspace ID strings
|
|
"""
|
|
query: dict[str, Any] = {"page": page, "size": size}
|
|
if reverse:
|
|
query["reverse"] = "true"
|
|
|
|
data = self._http.post(
|
|
routes.workspaces_list(),
|
|
body={"filters": filters} if filters else None,
|
|
query=query,
|
|
)
|
|
|
|
def transform(workspace: WorkspaceResponse) -> str:
|
|
"""Convert a workspace API response into its workspace ID."""
|
|
return workspace.id
|
|
|
|
def fetch_next(next_page: int) -> SyncPage[WorkspaceResponse, str]:
|
|
"""Fetch the next page while preserving filters and ordering."""
|
|
next_query: dict[str, Any] = {"page": next_page, "size": size}
|
|
if reverse:
|
|
next_query["reverse"] = "true"
|
|
next_data = self._http.post(
|
|
routes.workspaces_list(),
|
|
body={"filters": filters} if filters else None,
|
|
query=next_query,
|
|
)
|
|
return SyncPage(next_data, WorkspaceResponse, transform, fetch_next)
|
|
|
|
return SyncPage(data, WorkspaceResponse, transform, fetch_next)
|
|
|
|
@validate_call
|
|
def delete_workspace(
|
|
self,
|
|
workspace_id: str = Field(
|
|
..., min_length=1, description="ID of the workspace to delete"
|
|
),
|
|
) -> None:
|
|
"""
|
|
Delete a workspace.
|
|
|
|
Makes an API call to delete the specified workspace. This action cannot be undone.
|
|
|
|
Args:
|
|
workspace_id: The ID of the workspace to delete
|
|
"""
|
|
self._http.delete(routes.workspace(workspace_id))
|
|
|
|
@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
|
|
def chat(
|
|
self,
|
|
query: str = Field(..., min_length=1, description="The natural language query"),
|
|
*,
|
|
session: str | SessionBase | None = None,
|
|
reasoning_level: Literal["minimal", "low", "medium", "high", "max"]
|
|
| None = None,
|
|
response_format: type[BaseModel] | dict[str, Any] | None = None,
|
|
scope: str | list[str] | None = None,
|
|
) -> BaseModel | str | None:
|
|
"""
|
|
Query the entire workspace with a natural language question.
|
|
|
|
Unlike peer.chat(), which queries a single peer's representation, this
|
|
searches across ALL peers and observations in the workspace — use it
|
|
for cross-peer analysis, common themes, or workspace-wide questions.
|
|
|
|
Args:
|
|
query: The natural language question to ask.
|
|
session: Optional session to scope message retrieval to.
|
|
reasoning_level: Optional reasoning level: "minimal", "low",
|
|
"medium", "high", or "max" (default "low").
|
|
response_format: Optional structure for the answer: a Pydantic
|
|
model class (returns a parsed instance) or a raw
|
|
JSON Schema dict (returns a JSON string).
|
|
scope: Optional scope name(s) restricting recall to those scopes'
|
|
member sessions. Mutually exclusive with `session`.
|
|
|
|
Returns:
|
|
The synthesized answer, or None if no relevant information.
|
|
"""
|
|
self._ensure_workspace()
|
|
resolved_session_id = resolve_id(session)
|
|
body: dict[str, Any] = {"query": query, "stream": False}
|
|
if resolved_session_id:
|
|
body["session_id"] = resolved_session_id
|
|
if reasoning_level:
|
|
body["reasoning_level"] = reasoning_level
|
|
if scope is not None:
|
|
body["scope"] = scope
|
|
response_format_schema = serialize_response_format(response_format)
|
|
if response_format_schema is not None:
|
|
body["response_format"] = response_format_schema
|
|
|
|
data = self._http.post(
|
|
routes.workspace_chat(self.workspace_id),
|
|
body=body,
|
|
)
|
|
content = data.get("content")
|
|
if not content:
|
|
return None
|
|
if isinstance(response_format, type):
|
|
return response_format.model_validate_json(content)
|
|
return content
|
|
|
|
@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
|
|
def chat_stream(
|
|
self,
|
|
query: str = Field(..., min_length=1, description="The natural language query"),
|
|
*,
|
|
session: str | SessionBase | None = None,
|
|
reasoning_level: Literal["minimal", "low", "medium", "high", "max"]
|
|
| None = None,
|
|
response_format: type[BaseModel] | dict[str, Any] | None = None,
|
|
scope: str | list[str] | None = None,
|
|
) -> DialecticStreamResponse:
|
|
"""Streaming variant of :meth:`chat`. See chat() for argument docs."""
|
|
self._ensure_workspace()
|
|
resolved_session_id = resolve_id(session)
|
|
body: dict[str, Any] = {"query": query, "stream": True}
|
|
if resolved_session_id:
|
|
body["session_id"] = resolved_session_id
|
|
if reasoning_level:
|
|
body["reasoning_level"] = reasoning_level
|
|
if scope is not None:
|
|
body["scope"] = scope
|
|
response_format_schema = serialize_response_format(response_format)
|
|
if response_format_schema is not None:
|
|
body["response_format"] = response_format_schema
|
|
|
|
def stream_response() -> Generator[str, None, None]:
|
|
yield from parse_sse_stream(
|
|
self._http.stream(
|
|
"POST",
|
|
routes.workspace_chat(self.workspace_id),
|
|
body=body,
|
|
)
|
|
)
|
|
|
|
return DialecticStreamResponse(stream_response())
|
|
|
|
@validate_call
|
|
def search(
|
|
self,
|
|
query: str = Field(..., min_length=1, description="The search query to use"),
|
|
filters: dict[str, object] | None = Field(
|
|
None, description="Filters to scope the search"
|
|
),
|
|
limit: int = Field(
|
|
default=10, ge=1, le=100, description="Number of results to return"
|
|
),
|
|
*,
|
|
scope: str | ScopeBase | None = Field(
|
|
None,
|
|
description="Optional scope restricting the search to its member sessions",
|
|
),
|
|
) -> list[Message]:
|
|
"""
|
|
Search for messages in the current workspace.
|
|
|
|
Makes an API call to search for messages in the current workspace.
|
|
|
|
Args:
|
|
query: The search query to use
|
|
filters: Filters to scope the search. See [search filters documentation](https://honcho.dev/docs/v3/documentation/core-concepts/features/using-filters).
|
|
limit: Number of results to return (1-100, default: 10)
|
|
scope: Optional scope (ID or Scope object) restricting the search to
|
|
that scope's member sessions. Mutually exclusive with a
|
|
``session_id`` filter. A scope with no member sessions matches
|
|
nothing rather than everything.
|
|
|
|
Returns:
|
|
A list of Message objects representing the search results.
|
|
Returns an empty list if no messages are found.
|
|
"""
|
|
self._ensure_workspace()
|
|
body: dict[str, Any] = {"query": query, "filters": filters, "limit": limit}
|
|
if scope is not None:
|
|
body["scope"] = validate_scope_id(resolve_id(scope))
|
|
data = self._http.post(
|
|
routes.workspace_search(self.workspace_id),
|
|
body=body,
|
|
)
|
|
return [
|
|
Message.from_api_response(MessageResponse.model_validate(item))
|
|
for item in data
|
|
]
|
|
|
|
@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
|
|
def queue_status(
|
|
self,
|
|
observer: str | PeerBase | None = None,
|
|
sender: str | PeerBase | None = None,
|
|
session: str | SessionBase | None = None,
|
|
) -> QueueStatusResponse:
|
|
"""
|
|
Get the queue processing status, optionally scoped to an observer, sender, and/or session.
|
|
|
|
Args:
|
|
observer: Optional observer (ID string or Peer object) to scope the status check
|
|
sender: Optional sender (ID string or Peer object) to scope the status check
|
|
session: Optional session (ID string or Session object) to scope the status check
|
|
"""
|
|
self._ensure_workspace()
|
|
resolved_observer_id = resolve_id(observer)
|
|
resolved_sender_id = resolve_id(sender)
|
|
resolved_session_id = resolve_id(session)
|
|
|
|
query: dict[str, Any] = {}
|
|
if resolved_observer_id:
|
|
query["observer_id"] = resolved_observer_id
|
|
if resolved_sender_id:
|
|
query["sender_id"] = resolved_sender_id
|
|
if resolved_session_id:
|
|
query["session_id"] = resolved_session_id
|
|
|
|
data = self._http.get(
|
|
routes.workspace_queue_status(self.workspace_id),
|
|
query=query if query else None,
|
|
)
|
|
return QueueStatusResponse.model_validate(data)
|
|
|
|
@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
|
|
def schedule_dream(
|
|
self,
|
|
observer: str | PeerBase,
|
|
session: str | SessionBase | None = None,
|
|
observed: str | PeerBase | None = None,
|
|
) -> None:
|
|
"""
|
|
Schedule a dream task for memory consolidation.
|
|
|
|
Dreams are background processes that consolidate observations into higher-level
|
|
insights and update peer cards. This method schedules a dream task for immediate
|
|
processing.
|
|
|
|
Args:
|
|
observer: The observer peer (ID string or Peer object) whose perspective
|
|
to use for the dream.
|
|
session: Optional session (ID string or Session object) to scope the dream to.
|
|
observed: Optional observed peer (ID string or Peer object). If not provided,
|
|
defaults to the observer (self-reflection).
|
|
"""
|
|
self._ensure_workspace()
|
|
resolved_observer_id = resolve_id(observer)
|
|
resolved_session_id = resolve_id(session)
|
|
resolved_observed_id = (
|
|
resolve_id(observed) if observed else resolved_observer_id
|
|
)
|
|
|
|
self._http.post(
|
|
routes.workspace_schedule_dream(self.workspace_id),
|
|
body={
|
|
"observer": resolved_observer_id,
|
|
"observed": resolved_observed_id,
|
|
"session_id": resolved_session_id,
|
|
"dream_type": "omni",
|
|
},
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
"""
|
|
Return a string representation of the Honcho client.
|
|
|
|
Returns:
|
|
A string representation suitable for debugging
|
|
"""
|
|
return (
|
|
f"Honcho(workspace_id='{self.workspace_id}', base_url='{self._base_url}')"
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
"""
|
|
Return a human-readable string representation of the Honcho client.
|
|
|
|
Returns:
|
|
A string showing the workspace ID
|
|
"""
|
|
return f"Honcho Client (workspace: {self.workspace_id})"
|