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.")

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
    sessionId: str
    agentId: str
    turnIndex: int
    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  # "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  # e.g. "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, total=False):
    value: str
    displayName: str
    isEnabled: bool

class ModelPolicyTimeoutError(Exception): ...

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

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

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

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"]

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]]:
    ...

# 简单 dict：所有字段 required
{"query": str, "limit": int}

# 使用 Annotated 增加字段描述
{"query": Annotated[str, "Search keywords"]}

# TypedDict：支持 NotRequired
class SearchInput(TypedDict):
    query: str
    limit: NotRequired[int]

# 完整 JSON Schema dict
{
    "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],
]

@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

from typing import TypedDict

from typing_extensions import NotRequired


class McpToolAnnotations(TypedDict, total=False):
    readOnly: bool
    destructive: bool
    openWorld: bool


class McpToolInfo(TypedDict):
    name: str
    description: NotRequired[str]
    annotations: NotRequired[McpToolAnnotations]

HookEvent = Literal[
    "PreToolUse",
    "PostToolUse",
    "PostToolUseFailure",
    "UserPromptSubmit",
    "SessionStart",
    "SessionEnd",
    "Stop",
    "SubagentStart",
    "SubagentStop",
    "PreCompact",
    "PostCompact",
    "CwdChanged",
    "InstructionsLoaded",
    "FileChanged",
    "PermissionRequest",
]

HookCallback = Callable[
    [HookInput, str | None, HookCallbackOptions],
    Awaitable[HookJSONOutput],
]

@dataclass
class HookMatcher:
    hooks: list[HookCallback]
    matcher: str | None = None
    timeout: int | None = None  # 秒，默认 60

@dataclass
class BaseHookInput:
    hook_event_name: str
    session_id: str
    transcript_path: str
    cwd: str

@dataclass
class HookJSONOutput:
    continue_: bool | None = None       # JSON 键名为 "continue"
    stop_reason: str | None = None      # JSON 键名为 "stopReason"
    decision: str | None = None
    reason: str | None = None
    hook_specific_output: dict | None = None  # JSON 键名为 "hookSpecificOutput"

@dataclass
class PreToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PreToolUse"]
    permission_mode: str | None
    tool_name: str
    tool_input: Any

@dataclass
class PostToolUseHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUse"]
    tool_name: str
    tool_input: Any
    tool_response: Any

@dataclass
class PostToolUseFailureHookInput(BaseHookInput):
    hook_event_name: Literal["PostToolUseFailure"]
    tool_name: str
    tool_input: Any
    error: str
    is_interrupt: bool | None

@dataclass
class UserPromptSubmitHookInput(BaseHookInput):
    hook_event_name: Literal["UserPromptSubmit"]
    prompt: str

@dataclass
class SessionStartHookInput(BaseHookInput):
    hook_event_name: Literal["SessionStart"]
    source: str

@dataclass
class SessionEndHookInput(BaseHookInput):
    hook_event_name: Literal["SessionEnd"]
    reason: str

@dataclass
class StopHookInput(BaseHookInput):
    hook_event_name: Literal["Stop"]
    stop_hook_active: bool

@dataclass
class SubagentStartHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStart"]
    agent_id: str
    agent_type: str

@dataclass
class SubagentStopHookInput(BaseHookInput):
    hook_event_name: Literal["SubagentStop"]
    stop_hook_active: bool

@dataclass
class PreCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PreCompact"]
    trigger: str
    custom_instructions: str | None

@dataclass
class PostCompactHookInput(BaseHookInput):
    hook_event_name: Literal["PostCompact"]
    trigger: str
    compact_summary: str

@dataclass
class CwdChangedHookInput(BaseHookInput):
    hook_event_name: Literal["CwdChanged"]
    old_cwd: str
    new_cwd: str

@dataclass
class InstructionsLoadedHookInput(BaseHookInput):
    hook_event_name: Literal["InstructionsLoaded"]
    load_reason: str

@dataclass
class FileChangedHookInput(BaseHookInput):
    hook_event_name: Literal["FileChanged"]
    file_path: str
    event: str

@dataclass
class PermissionRequestHookInput(BaseHookInput):
    hook_event_name: Literal["PermissionRequest"]
    tool_name: str
    tool_input: Any
    permission_suggestions: list | None

@dataclass
class AssistantMessage:
    content: list[TextBlock | ToolUseBlock]
    parent_tool_use_id: str | None = None
    session_id: str | None = None
    uuid: str | None = None

@dataclass
class ResultMessage:
    subtype: str  # "success" | "error_max_turns" | "error_during_execution" | ...
    duration_ms: int
    num_turns: int
    session_id: str
    total_cost_usd: float | None = None
    result: str | None = None  # 仅在 subtype == "success" 时存在

@dataclass
class SystemMessage:
    subtype: str  # "init" | "compact_boundary" | "status" | "mcp_status_change" | ...
    data: dict[str, Any]

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]  # 上游兼容的原始 stream event
    parent_tool_use_id: str | None = None