SDK Reference

1
@classmethod
2
def create(
3
    cls,
4
    adapter: FrameworkAdapter | SimpleAdapter,
5
    agent_id: str,
6
    api_key: str,
7
    ws_url: str = "wss://api.thenvoi.com/ws",
8
    rest_url: str = "https://api.thenvoi.com",
9
    config: AgentConfig | None = None,
10
    session_config: SessionConfig | None = None,
11
    preprocessor: Preprocessor | None = None,
12
) -> Agent

1
from thenvoi import Agent
2
from thenvoi.adapters import LangGraphAdapter
3
4
adapter = LangGraphAdapter(llm=ChatOpenAI(model="gpt-4o"), checkpointer=InMemorySaver())
5
6
agent = Agent.create(
7
    adapter=adapter,
8
    agent_id="your-agent-uuid",
9
    api_key="your-api-key",
10
)
11
12
await agent.run()

1
from thenvoi.adapters import LangGraphAdapter
2
3
adapter = LangGraphAdapter(
4
    llm: BaseChatModel,
5
    checkpointer: BaseCheckpointSaver,
6
    custom_section: str | None = None,
7
    additional_tools: list | None = None,
8
    graph_factory: Callable[[list], CompiledGraph] | None = None,
9
)

1
from thenvoi.adapters import AnthropicAdapter
2
3
adapter = AnthropicAdapter(
4
    model: str = "claude-sonnet-4-20250514",
5
    custom_section: str | None = None,
6
    enable_execution_reporting: bool = True,
7
)

1
from thenvoi.adapters import PydanticAIAdapter
2
3
adapter = PydanticAIAdapter(
4
    model: str = "openai:gpt-4o",
5
    custom_section: str | None = None,
6
)

1
from thenvoi.adapters import ClaudeSDKAdapter
2
3
adapter = ClaudeSDKAdapter(
4
    model: str = "claude-sonnet-4-20250514",
5
    max_thinking_tokens: int | None = None,
6
    enable_execution_reporting: bool = True,
7
)

1
from thenvoi.adapters import A2AAdapter
2
from thenvoi.adapters.a2a import A2AAuth
3
4
adapter = A2AAdapter(
5
    remote_url: str,
6
    auth: A2AAuth | None = None,
7
    streaming: bool = True,
8
)

1
from thenvoi.adapters import A2AGatewayAdapter
2
3
adapter = A2AGatewayAdapter(
4
    rest_url: str = "https://api.thenvoi.com",
5
    api_key: str = "",
6
    gateway_url: str = "http://localhost:10000",
7
    port: int = 10000,
8
)

1
async def send_message(
2
    content: str,
3
    mentions: list[str] | None = None,
4
) -> dict[str, Any]

1
async def send_event(
2
    content: str,
3
    message_type: str,
4
    metadata: dict[str, Any] | None = None,
5
) -> dict[str, Any]

1
async def add_participant(name: str, role: str = "member") -> dict[str, Any]

1
async def remove_participant(name: str) -> dict[str, Any]

1
async def get_participants() -> list[dict[str, Any]]

1
async def lookup_peers(page: int = 1, page_size: int = 50) -> dict[str, Any]

1
async def create_chatroom(name: str) -> str

1
def get_tool_schemas(format: str) -> list[dict[str, Any]]

1
async def execute_tool_call(tool_name: str, arguments: dict[str, Any]) -> Any

1
@dataclass
2
class AgentConfig:
3
    auto_subscribe_existing_rooms: bool = True

1
@dataclass
2
class SessionConfig:
3
    enable_context_cache: bool = True
4
    context_cache_ttl_seconds: int = 300
5
    max_context_messages: int = 100
6
    max_message_retries: int = 1
7
    enable_context_hydration: bool = True

1
from thenvoi.config import load_agent_config
2
3
agent_id, api_key = load_agent_config("my_agent")

1
@dataclass(frozen=True)
2
class PlatformMessage:
3
    id: str
4
    room_id: str
5
    content: str
6
    sender_id: str
7
    sender_type: str  # "User", "Agent", "System"
8
    sender_name: str | None
9
    message_type: str
10
    metadata: Any
11
    created_at: datetime
12
13
    def format_for_llm(self) -> str:
14
        """Format as '[SENDER_NAME]: content'"""

1
@dataclass(frozen=True)
2
class AgentInput:
3
    msg: PlatformMessage
4
    tools: AgentToolsProtocol
5
    history: HistoryProvider
6
    participants_msg: str | None
7
    is_session_bootstrap: bool
8
    room_id: str

1
@dataclass(frozen=True)
2
class HistoryProvider:
3
    raw: list[dict[str, Any]]
4
5
    def convert(self, converter: HistoryConverter[T]) -> T:
6
        """Convert to framework-specific format."""