SDK References

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: QoderAgentOptions | None = None,
) -> AsyncIterator[Message]:
    ...

client = QoderSDKClient(options=options)
await client.connect("Initial prompt.")

async def query(
    prompt: str | AsyncIterable[dict[str, Any]],
    session_id: str = "default",
    *,
    priority: Literal["now", "next", "later"] = "next",
    message_uuid: str | None = None,
    should_query: bool = True,
) -> None:
    ...

async def interrupt() -> None:
    ...

async def cancel_async_message(message_uuid: str) -> bool:
    ...

async def get_server_info() -> SDKControlInitializeResponse | None:
    ...

async def list_plugins() -> list[PluginDetails]:
    ...

async def list_sessions_from_store(store, directory=None, limit=None, offset=0) -> list[SDKSessionInfo]: ...
async def get_session_info_from_store(store, session_id, directory=None) -> SDKSessionInfo | None: ...
async def get_session_messages_from_store(store, session_id, directory=None, limit=None, offset=0) -> list[SessionMessage]: ...
async def list_subagents_from_store(store, session_id, directory=None) -> list[str]: ...
async def get_subagent_messages_from_store(store, session_id, agent_id, directory=None, limit=None, offset=0) -> list[SessionMessage]: ...
async def rename_session_via_store(store, session_id, title, directory=None) -> None: ...
async def tag_session_via_store(store, session_id, tag, directory=None) -> None: ...
async def delete_session_via_store(store, session_id, directory=None) -> None: ...
async def fork_session_via_store(store, session_id, directory=None, up_to_message_id=None, title=None) -> ForkSessionResult: ...

async def import_session_to_store(
    session_id: str,
    store: SessionStore,
    *,
    directory: str | Path | None = None,
    include_subagents: bool = True,
    batch_size: int = 500,
) -> None:
    ...

class SessionStore(Protocol):
    async def append(
        self, key: SessionKey, entries: list[SessionStoreEntry]
    ) -> None: ...

    async def load(
        self, key: SessionKey
    ) -> list[SessionStoreEntry] | None: ...

    async def list_sessions(
        self, project_key: str
    ) -> list[SessionStoreListEntry]: ...

    async def delete(self, key: SessionKey) -> None: ...

    async def list_subkeys(
        self, key: SessionListSubkeysKey
    ) -> list[str]: ...

SessionStoreFlushMode = Literal["batched", "eager"]

@dataclass
class SDKSessionInfo:
    session_id: str
    summary: str
    last_modified: int
    file_size: int | None = None
    custom_title: str | None = None
    first_prompt: str | None = None
    git_branch: str | None = None
    cwd: str | None = None
    tag: str | None = None
    created_at: int | None = None

@dataclass
class SessionMessage:
    type: Literal["user", "assistant"]
    uuid: str
    session_id: str
    message: Any
    parent_tool_use_id: str | None = None

settings = {
    "skillOverrides": {
        "plugin:review": "off",
    },
    "skillListingMaxDescChars": 1200,
    "skillListingBudgetFraction": 0.01,
}

class SDKControlInitializeResponse(TypedDict):
    commands: list[InitializeCommandInfo]
    agents: list[InitializeAgentInfo]
    skills: NotRequired[list[InitializeSkillInfo]]
    output_style: str
    available_output_styles: list[str]
    models: list[ModelInfo]
    account: dict[str, Any]
    fast_mode_state: NotRequired[dict[str, Any]]

class InitializeSkillInfo(TypedDict):
    name: str
    description: NotRequired[str]
    source: NotRequired[str]

from qoder_agent_sdk import access_token, access_token_from_env, qodercli_auth

from qoder_agent_sdk import AgentDefinition, QoderAgentOptions


options = QoderAgentOptions(
    allowed_tools=["Agent"],
    agents={
        "reviewer": AgentDefinition(
            description="Reviews code quality and reports actionable findings.",
            prompt="Review the requested code and report concrete issues.",
            tools=["Read", "Grep", "Glob"],
        ),
    },
)

options = QoderAgentOptions(
    agents={
        "planner": AgentDefinition(
            description="Plans work before implementation.",
            prompt="Break work into steps, risks, and validation checks.",
            tools=["Read", "Grep", "Glob"],
        ),
    },
    agent="planner",
)

from dataclasses import dataclass
from typing import Any, Literal


@dataclass
class AgentDefinition:
    description: str
    prompt: str
    tools: list[str] | None = None
    disallowedTools: list[str] | None = None
    model: str | None = None
    skills: list[str] | None = None
    mcpServers: list[str | dict[str, Any]] | None = None
    initialPrompt: str | None = None
    maxTurns: int | None = None
    effort: Literal["low", "medium", "high", "max"] | None = None
    permissionMode: PermissionMode | None = None

description="Runs project tests, analyzes failing output, and suggests fixes."

prompt="""You are a security reviewer.
Check for authentication bypass, authorization bugs, injection risks, and secret leaks.
Return findings sorted by severity."""

tools=["Read", "Grep", "Glob"]

disallowedTools=["Bash", "Write"]

mcpServers: list[str | dict[str, Any]] | None

options = QoderAgentOptions(
    mcp_servers={
        "orders": {
            "command": "python",
            "args": ["servers/orders.py"],
        },
    },
    allowed_tools=["Agent"],
    agents={
        "support": AgentDefinition(
            description="Answers support questions using order tools.",
            prompt="Use order tools when needed and return a concise answer.",
            mcpServers=["orders"],
            tools=["mcp__orders__lookup_order"],
        ),
    },
)

AgentDefinition(
    description="Searches the internal knowledge base.",
    prompt="Search the knowledge base and cite relevant entries.",
    mcpServers=[
        {
            "kb": {
                "command": "python",
                "args": ["servers/kb.py"],
            },
        },
    ],
    tools=["mcp__kb__search"],
)

skills=["review", "sdk-test-plugin:sdk-echo"]

initialPrompt="Start by scanning authentication and session management code."

maxTurns=6

EffortLevel = Literal["low", "medium", "high", "max"]

PermissionMode = Literal[
    "default",
    "acceptEdits",
    "plan",
    "bypassPermissions",
    "yolo",
    "dontAsk",
    "auto",
]

class AgentInfo(TypedDict):
    name: str
    description: str
    model: NotRequired[str | None]

from qoder_agent_sdk import AgentDefinition, QoderAgentOptions, QoderSDKClient


options = QoderAgentOptions(
    agents={
        "reviewer": AgentDefinition(
            description="Reviews code quality.",
            prompt="Review code and report findings.",
        ),
    },
)

client = QoderSDKClient(options=options)
await client.connect("List agents.")
agents = client.supported_agents()
await client.disconnect()

from typing import Awaitable, Callable

ModelPolicyProvider = Callable[
    ["ModelPolicyContext"],
    "ModelPolicyResult | Awaitable[ModelPolicyResult]",
]

class ModelPolicyContext(TypedDict, total=False):
    purpose: str  # QoderModelPurpose type
    sessionId: str
    availableModels: list[ModelInfo]

from typing import Literal

QoderModelPurpose = Literal[
    "main",
    "plan",
    "task",
    "compact",
    "title",
    "suggestion",
    "generate",
    "hook_prompt",
    "subagent",
    "web_fetch",
    "image_gen",
    "compression",
    "utility",
]

class ModelPolicyResult(TypedDict, total=False):
    model: str | CustomModel
    parameters: dict[str, Any]

class CustomModel(TypedDict, total=False):
    provider: str
    model: str
    api_key: str
    url: str
    style: str  # Allowed values: "openai" | "anthropic"

class SDKControlGetByokConfigResponse(TypedDict, total=False):
    providers: list[BYOKProviderInfo]


class BYOKProviderInfo(TypedDict, total=False):
    key: str
    display_name: str
    api_key_url: str
    url: str
    fields: list[BYOKFieldInfo]
    types: list[BYOKModelTypeInfo]


class BYOKFieldInfo(TypedDict, total=False):
    key: str
    display_name: str
    type: str  # For example, "free_input"
    mandatory: bool


class BYOKModelTypeInfo(TypedDict, total=False):
    key: str
    display_name: str
    models: list[BYOKModelInfo]


class BYOKModelInfo(TypedDict, total=False):
    key: str
    display_name: str
    is_vl: bool
    is_reasoning: bool
    format: str
    max_input_tokens: int

class ModelInfo(TypedDict):
    value: str
    displayName: str
    description: str
    isEnabled: bool
    isNew: NotRequired[bool]
    isFree: NotRequired[bool]
    priceFactor: NotRequired[float]
    context_config: NotRequired[ModelContextConfig]
    thinking_config: NotRequired[ModelThinkingConfig]
    promotion: NotRequired[ModelPromotion]
    serverModel: NotRequired[ServerModelJson]

ModelContextConfig = dict[str, ModelContextWindowEntry]

class ModelContextWindowEntry(TypedDict):
    token_count: int
    is_default: NotRequired[bool]

class ModelThinkingConfig(TypedDict, total=False):
    disabled: ModelThinkingDisabled
    enabled: ModelThinkingEnabled

class ModelThinkingDisabled(TypedDict, total=False):
    description: str

class ModelThinkingEnabled(TypedDict, total=False):
    description: str
    efforts: dict[str, ModelEffortEntry]
    is_default: bool

class ModelEffortEntry(TypedDict, total=False):
    description: str
    is_default: bool

LocalizedModelText = dict[str, str]

class ModelPromotion(TypedDict, total=False):
    active: bool
    badge: LocalizedModelText
    description: LocalizedModelText
    discount_factor: float
    before_promotion_price_factor: float
    timezone: str
    rule_id: str
    window_start: str
    window_end: str

ServerModelJson = dict[str, ServerModelJsonValue]

class UsageQuotaBucket(TypedDict, total=False):
    total: float
    used: float
    remaining: float
    percentage: float
    unit: str

class UsageAddOnQuotaBucket(UsageQuotaBucket, total=False):
    detailUrl: str

class UsageOrgResourcePackage(TypedDict, total=False):
    used: float
    cap: float
    remaining: float
    percentage: float
    available: bool
    unit: str

class UsageInfo(TypedDict, total=False):
    userId: str
    userType: str
    totalUsagePercentage: float
    isHighestTier: bool
    expiresAt: float
    upgradeUrl: NotRequired[str]
    userQuota: UsageQuotaBucket
    addOnQuota: NotRequired[UsageAddOnQuotaBucket]
    isQuotaExceeded: bool
    isPlanQuotaProrated: bool
    orgResourcePackage: UsageOrgResourcePackage

class ModelPolicyTimeoutError(Exception): ...

async def set_model(self, model: str | None = None) -> None: ...

async def set_proxy(self, proxy: str | None = None) -> None: ...

async def get_available_models(self) -> list[ModelInfo]: ...

async def list_byok_providers(self) -> list[BYOKProviderInfo] | None: ...

async def get_usage_info(self) -> UsageInfo | None: ...

async with QoderSDKClient() as client:
    usage = await client.get_usage_info()
    if usage is not None:
        print(f"Plan: {usage['userType']}, used {usage['totalUsagePercentage']}%")
        user_quota = usage.get("userQuota") or {}
        print(f"Plan quota: {user_quota.get('remaining')}/{user_quota.get('total')} {user_quota.get('unit')} left")

from collections.abc import Awaitable, Callable
from typing import Any


CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext],
    Awaitable[PermissionResult],
]

import asyncio
from dataclasses import dataclass, field
from typing import Any


@dataclass
class ToolPermissionContext:
    signal: asyncio.Event | None = None
    suggestions: list[Any] = field(default_factory=list)
    blocked_path: str | None = None
    decision_reason: str | None = None
    decision_reason_type: str | None = None
    classifier_approvable: bool | None = None
    title: str | None = None
    display_name: str | None = None
    description: str | None = None
    tool_use_id: str | None = None
    agent_id: str | None = None
    exit_plan_mode: ExitPlanModeApprovalDetails | None = None

PermissionMode = Literal[
    "default",
    "acceptEdits",
    "bypassPermissions",
    "yolo",
    "plan",
    "dontAsk",
    "auto",
]

from dataclasses import dataclass
from typing import Any, Literal


@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    updated_permissions: list[PermissionUpdate | dict[str, Any]] | None = None
    decision_classification: PermissionDecisionClassification | None = None


@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False
    decision_classification: PermissionDecisionClassification | None = None


PermissionResult = PermissionResultAllow | PermissionResultDeny

McpServerConfig = (
    McpStdioServerConfig
    | McpSSEServerConfig
    | McpHttpServerConfig
    | McpSdkServerConfig
)

class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]
    tools: NotRequired[list[McpServerToolPolicy]]

class McpSSEServerConfig(TypedDict):
    type: Literal["sse"]
    url: str
    headers: NotRequired[dict[str, str]]
    tools: NotRequired[list[McpServerToolPolicy]]

class McpHttpServerConfig(TypedDict):
    type: Literal["http"]
    url: str
    headers: NotRequired[dict[str, str]]
    tools: NotRequired[list[McpServerToolPolicy]]

class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: McpServer
    tools: NotRequired[list[McpServerToolPolicy]]

class McpServerToolPolicy(TypedDict):
    name: str
    permission_policy: Literal["always_allow", "always_ask", "always_deny"]

class SdkPluginConfig(TypedDict):
    type: Literal["local"]
    path: str

SettingSource = Literal["user", "project", "local"]

class CloudAgentOptions(TypedDict, total=False):
    agent: CloudAgentReference
    session: CloudSessionReference
    stream: CloudAgentStreamOptions

class AgentCreateParams(TypedDict, total=False):
    model: str
    name: str
    description: str | None
    system: str | None
    tools: list[BetaManagedAgentsAgentToolset20260401Params]
    mcp_servers: list[BetaManagedAgentsURLMCPServerParams]
    skills: list[BetaManagedAgentsSkillParams]
    metadata: dict[str, str]

class BetaManagedAgentsAgentToolset20260401Params(TypedDict, total=False):
    type: Literal["agent_toolset_20260401"]
    enabled_tools: list[BetaManagedAgentsAgentToolName]


BetaManagedAgentsAgentToolName = Literal[
    "bash", "write", "glob", "web_fetch", "read", "edit", "grep", "web_search"
]

class BetaManagedAgentsURLMCPServerParams(TypedDict):
    name: str
    type: Literal["url"]
    url: str

class BetaManagedAgentsCustomSkillParams(TypedDict):
    skill_id: str
    type: Literal["custom"]

class CloudSessionCreateParams(TypedDict, total=False):
    environment_id: str
    resources: list[CloudSessionResource]
    title: str | None
    vault_ids: list[str]
    memory_store_ids: list[str]

class BetaManagedAgentsFileResourceParams(TypedDict, total=False):
    file_id: str
    type: Literal["file"]
    path: str

CloudSessionResource = BetaManagedAgentsFileResourceParams

class CloudAgentStreamOptions(TypedDict, total=False):
    after_id: str
    delta_flush_interval_ms: int
    # Compatibility aliases
    afterId: str
    deltaFlushIntervalMs: int

@dataclass
class CloudAgentEventMessage:
    event: str
    data: Any
    uuid: str
    session_id: str
    id: str | None = None

from qoder_agent_sdk import CloudAgentApiError, CloudAgentUnsupportedAuthError

type ToolConfig = {
  askUserQuestion?: {
    previewFormat?: "markdown" | "html";
  };
};

mcp__{serverName}__{toolName}

from collections.abc import Awaitable, Callable
from typing import Any

from mcp.types import ToolAnnotations


def tool(
    name: str,
    description: str,
    input_schema: type | dict[str, Any],
    annotations: ToolAnnotations | None = None,
) -> Callable[[Callable[..., Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]:
    ...

# Plain dictionary: all fields are required
{"query": str, "limit": int}

# Use Annotated to add field descriptions
{"query": Annotated[str, "Search keywords"]}

# TypedDict: supports NotRequired
class SearchInput(TypedDict):
    query: str
    limit: NotRequired[int]

# Complete JSON Schema dictionary
{
    "type": "object",
    "properties": {
        "query": {"type": "string"},
        "source": {"type": "string", "enum": ["docs", "wiki"]},
    },
    "required": ["query"],
}

from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Generic, TypeVar

from mcp.types import ToolAnnotations


T = TypeVar("T")


@dataclass
class SdkMcpTool(Generic[T]):
    name: str
    description: str
    input_schema: type[T] | dict[str, Any]
    handler: Callable[..., Awaitable[dict[str, Any]]]
    annotations: ToolAnnotations | None = None

import asyncio
from dataclasses import dataclass, field


@dataclass
class ToolInvocationContext:
    signal: asyncio.Event = field(default_factory=asyncio.Event)

@tool("watch", "Watch a counter.", {"max": int})
async def watch(args, extra: ToolInvocationContext):
    ...

from mcp.types import ToolAnnotations


ToolAnnotations(
    title="Search docs",
    readOnlyHint=True,
    destructiveHint=False,
    idempotentHint=True,
    openWorldHint=False,
    maxResultSizeChars=500_000,
)

from typing import Any


def create_sdk_mcp_server(
    name: str,
    version: str = "1.0.0",
    tools: list[SdkMcpTool[Any]] | None = None,
) -> McpSdkServerConfig:
    ...

from typing import Literal, TypedDict

from typing_extensions import NotRequired


class McpSdkServerConfig(TypedDict):
    type: Literal["sdk"]
    name: str
    instance: McpServer
    tools: NotRequired[list[McpServerToolPolicy]]

class McpServerToolPolicy(TypedDict):
    name: str
    permission_policy: Literal["always_allow", "always_ask", "always_deny"]

from typing import Literal, TypedDict

from typing_extensions import NotRequired


class TextToolContent(TypedDict):
    type: Literal["text"]
    text: str


class ImageToolContent(TypedDict):
    type: Literal["image"]
    data: str
    mimeType: str


class ResourceLinkToolContent(TypedDict):
    type: Literal["resource_link"]
    uri: str
    name: NotRequired[str]
    description: NotRequired[str]
    mimeType: NotRequired[str]


class EmbeddedResourceValue(TypedDict, total=False):
    uri: str
    mimeType: str
    text: str
    blob: str


class EmbeddedResourceToolContent(TypedDict):
    type: Literal["resource"]
    resource: EmbeddedResourceValue


ToolContent = (
    TextToolContent
    | ImageToolContent
    | ResourceLinkToolContent
    | EmbeddedResourceToolContent
)


class ToolHandlerResult(TypedDict):
    content: list[ToolContent]
    is_error: NotRequired[bool]

from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from typing import Any, Literal


@dataclass
class ToolPermissionContext:
    signal: asyncio.Event | None = None
    suggestions: list[Any] = field(default_factory=list)
    blocked_path: str | None = None
    decision_reason: str | None = None
    decision_reason_type: str | None = None
    classifier_approvable: bool | None = None
    title: str | None = None
    display_name: str | None = None
    description: str | None = None
    tool_use_id: str | None = None
    agent_id: str | None = None
    exit_plan_mode: ExitPlanModeApprovalDetails | None = None


CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext],
    Awaitable[PermissionResult],
]