"""Claude SDK for Python.""" import logging import sys import types as builtin_types from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import Annotated, Any, Generic, TypeVar, Union, get_args, get_origin if sys.version_info >= (3, 11): from typing import get_type_hints as _get_type_hints from typing import is_typeddict else: # On 3.10, stdlib is_typeddict doesn't recognize typing_extensions.TypedDict # subclasses, and stdlib get_type_hints doesn't strip NotRequired markers. from typing_extensions import get_type_hints as _get_type_hints from typing_extensions import is_typeddict from mcp.types import ToolAnnotations from ._errors import ( ClaudeSDKError, CLIConnectionError, CLIJSONDecodeError, CLINotFoundError, ProcessError, ) from ._internal.session_import import import_session_to_store from ._internal.session_mutations import ( ForkSessionResult, delete_session, delete_session_via_store, fork_session, fork_session_via_store, rename_session, rename_session_via_store, tag_session, tag_session_via_store, ) from ._internal.session_store import InMemorySessionStore, project_key_for_directory from ._internal.session_summary import fold_session_summary from ._internal.sessions import ( get_session_info, get_session_info_from_store, get_session_messages, get_session_messages_from_store, get_subagent_messages, get_subagent_messages_from_store, list_sessions, list_sessions_from_store, list_subagents, list_subagents_from_store, ) from ._internal.transport import Transport from ._version import __version__ from .client import ClaudeSDKClient from .query import query from .types import ( TERMINAL_TASK_STATUSES, AgentDefinition, AssistantMessage, BaseHookInput, CanUseTool, ClaudeAgentOptions, ContentBlock, ContextUsageCategory, ContextUsageResponse, DeferredToolUse, EffortLevel, HookCallback, HookContext, HookEventMessage, HookInput, HookJSONOutput, HookMatcher, McpSdkServerConfig, McpServerConfig, McpServerConnectionStatus, McpServerInfo, McpServerStatus, McpServerStatusConfig, McpStatusResponse, McpToolAnnotations, McpToolInfo, Message, MirrorErrorMessage, NotificationHookInput, NotificationHookSpecificOutput, PermissionMode, PermissionRequestHookInput, PermissionRequestHookSpecificOutput, PermissionResult, PermissionResultAllow, PermissionResultDeny, PermissionUpdate, PostToolUseFailureHookInput, PostToolUseFailureHookSpecificOutput, PostToolUseHookInput, PreCompactHookInput, PreToolUseHookInput, RateLimitEvent, RateLimitInfo, RateLimitStatus, RateLimitType, ResultMessage, SandboxIgnoreViolations, SandboxNetworkConfig, SandboxSettings, SdkBeta, SdkPluginConfig, SDKSessionInfo, ServerToolName, ServerToolResultBlock, ServerToolUseBlock, SessionKey, SessionListSubkeysKey, SessionMessage, SessionStore, SessionStoreEntry, SessionStoreFlushMode, SessionStoreListEntry, SessionSummaryEntry, SettingSource, StopHookInput, StreamEvent, SubagentStartHookInput, SubagentStartHookSpecificOutput, SubagentStopHookInput, SystemMessage, TaskBudget, TaskNotificationMessage, TaskNotificationStatus, TaskProgressMessage, TaskStartedMessage, TaskUpdatedMessage, TaskUpdatedStatus, TaskUsage, TextBlock, ThinkingBlock, ThinkingConfig, ThinkingConfigAdaptive, ThinkingConfigDisabled, ThinkingConfigEnabled, ToolPermissionContext, ToolResultBlock, ToolUseBlock, UserMessage, UserPromptSubmitHookInput, ) # MCP Server Support logger = logging.getLogger(__name__) T = TypeVar("T") @dataclass class SdkMcpTool(Generic[T]): """Definition for an SDK MCP tool.""" name: str description: str input_schema: type[T] | dict[str, Any] handler: Callable[[T], Awaitable[dict[str, Any]]] annotations: ToolAnnotations | None = None def tool( name: str, description: str, input_schema: type | dict[str, Any], annotations: ToolAnnotations | None = None, ) -> Callable[[Callable[[Any], Awaitable[dict[str, Any]]]], SdkMcpTool[Any]]: """Decorator for defining MCP tools with type safety. Creates a tool that can be used with SDK MCP servers. The tool runs in-process within your Python application, providing better performance than external MCP servers. Args: name: Unique identifier for the tool. This is what Claude will use to reference the tool in function calls. description: Human-readable description of what the tool does. This helps Claude understand when to use the tool. input_schema: Schema defining the tool's input parameters. Can be either: - A dictionary mapping parameter names to types (e.g., {"text": str}) - A TypedDict class for more complex schemas - A JSON Schema dictionary for full validation Use ``Annotated[type, "description"]`` to add a description to a parameter in either dict-style or TypedDict schemas. Returns: A decorator function that wraps the tool implementation and returns an SdkMcpTool instance ready for use with create_sdk_mcp_server(). Example: Basic tool with simple schema: >>> @tool("greet", "Greet a user", {"name": str}) ... async def greet(args): ... return {"content": [{"type": "text", "text": f"Hello, {args['name']}!"}]} Tool with multiple parameters: >>> @tool("add", "Add two numbers", {"a": float, "b": float}) ... async def add_numbers(args): ... result = args["a"] + args["b"] ... return {"content": [{"type": "text", "text": f"Result: {result}"}]} Tool with error handling: >>> @tool("divide", "Divide two numbers", {"a": float, "b": float}) ... async def divide(args): ... if args["b"] == 0: ... return {"content": [{"type": "text", "text": "Error: Division by zero"}], "is_error": True} ... return {"content": [{"type": "text", "text": f"Result: {args['a'] / args['b']}"}]} Notes: - The tool function must be async (defined with async def) - The function receives a single dict argument with the input parameters - The function should return a dict with a "content" key containing the response - Errors can be indicated by including "is_error": True in the response """ def decorator( handler: Callable[[Any], Awaitable[dict[str, Any]]], ) -> SdkMcpTool[Any]: return SdkMcpTool( name=name, description=description, input_schema=input_schema, handler=handler, annotations=annotations, ) return decorator def _python_type_to_json_schema(py_type: Any) -> dict[str, Any]: """Convert a Python type annotation to a JSON Schema dict.""" origin = get_origin(py_type) # NotRequired/Required/ReadOnly survive include_extras=True; unwrap them if getattr(origin, "_name", None) in ("NotRequired", "Required", "ReadOnly"): return _python_type_to_json_schema(get_args(py_type)[0]) if origin is Annotated: args = get_args(py_type) schema = _python_type_to_json_schema(args[0]) for meta in args[1:]: if isinstance(meta, str): schema["description"] = meta break return schema if py_type is str: return {"type": "string"} if py_type is int: return {"type": "integer"} if py_type is float: return {"type": "number"} if py_type is bool: return {"type": "boolean"} origin = getattr(py_type, "__origin__", None) if origin is Union or isinstance(py_type, builtin_types.UnionType): args = py_type.__args__ non_none = [a for a in args if a is not builtin_types.NoneType] if len(non_none) == 1: return _python_type_to_json_schema(non_none[0]) return {"anyOf": [_python_type_to_json_schema(a) for a in non_none]} if origin is list: item_args = getattr(py_type, "__args__", None) if item_args: return {"type": "array", "items": _python_type_to_json_schema(item_args[0])} return {"type": "array"} if origin is dict: return {"type": "object"} if py_type is list: return {"type": "array"} if py_type is dict: return {"type": "object"} if is_typeddict(py_type): return _typeddict_to_json_schema(py_type) return {"type": "string"} def _typeddict_to_json_schema(td_class: type) -> dict[str, Any]: """Convert a TypedDict class to a JSON Schema dict.""" hints = _get_type_hints(td_class, include_extras=True) properties: dict[str, Any] = {} for field_name, field_type in hints.items(): properties[field_name] = _python_type_to_json_schema(field_type) required_keys = getattr(td_class, "__required_keys__", set(properties.keys())) schema: dict[str, Any] = { "type": "object", "properties": properties, } if required_keys: schema["required"] = sorted(required_keys) return schema def create_sdk_mcp_server( name: str, version: str = "1.0.0", tools: list[SdkMcpTool[Any]] | None = None ) -> McpSdkServerConfig: """Create an in-process MCP server that runs within your Python application. Unlike external MCP servers that run as separate processes, SDK MCP servers run directly in your application's process. This provides: - Better performance (no IPC overhead) - Simpler deployment (single process) - Easier debugging (same process) - Direct access to your application's state Args: name: Unique identifier for the server. This name is used to reference the server in the mcp_servers configuration. version: Server version string. Defaults to "1.0.0". This is for informational purposes and doesn't affect functionality. tools: List of SdkMcpTool instances created with the @tool decorator. These are the functions that Claude can call through this server. If None or empty, the server will have no tools (rarely useful). Returns: McpSdkServerConfig: A configuration object that can be passed to ClaudeAgentOptions.mcp_servers. This config contains the server instance and metadata needed for the SDK to route tool calls. Example: Simple calculator server: >>> @tool("add", "Add numbers", {"a": float, "b": float}) ... async def add(args): ... return {"content": [{"type": "text", "text": f"Sum: {args['a'] + args['b']}"}]} >>> >>> @tool("multiply", "Multiply numbers", {"a": float, "b": float}) ... async def multiply(args): ... return {"content": [{"type": "text", "text": f"Product: {args['a'] * args['b']}"}]} >>> >>> calculator = create_sdk_mcp_server( ... name="calculator", ... version="2.0.0", ... tools=[add, multiply] ... ) >>> >>> # Use with Claude >>> options = ClaudeAgentOptions( ... mcp_servers={"calc": calculator}, ... allowed_tools=["add", "multiply"] ... ) Server with application state access: >>> class DataStore: ... def __init__(self): ... self.items = [] ... >>> store = DataStore() >>> >>> @tool("add_item", "Add item to store", {"item": str}) ... async def add_item(args): ... store.items.append(args["item"]) ... return {"content": [{"type": "text", "text": f"Added: {args['item']}"}]} >>> >>> server = create_sdk_mcp_server("store", tools=[add_item]) Notes: - The server runs in the same process as your Python application - Tools have direct access to your application's variables and state - No subprocess or IPC overhead for tool calls - Server lifecycle is managed automatically by the SDK See Also: - tool(): Decorator for creating tool functions - ClaudeAgentOptions: Configuration for using servers with query() """ from mcp.server import Server from mcp.types import ( AudioContent, CallToolResult, EmbeddedResource, ImageContent, ResourceLink, TextContent, Tool, ) # Create MCP server instance server = Server(name, version=version) # Register tools if provided if tools: # Store tools for access in handlers tool_map = {tool_def.name: tool_def for tool_def in tools} # Pre-compute tool schemas once at creation time def _build_schema(tool_def: SdkMcpTool[Any]) -> dict[str, Any]: if isinstance(tool_def.input_schema, dict): if ( "type" in tool_def.input_schema and "properties" in tool_def.input_schema and isinstance(tool_def.input_schema["type"], str) ): return tool_def.input_schema properties = {} for param_name, param_type in tool_def.input_schema.items(): properties[param_name] = _python_type_to_json_schema(param_type) return { "type": "object", "properties": properties, "required": list(properties.keys()), } if is_typeddict(tool_def.input_schema): return _typeddict_to_json_schema(tool_def.input_schema) return {"type": "object", "properties": {}} def _build_meta(tool_def: "SdkMcpTool[Any]") -> dict[str, Any] | None: # The MCP SDK's Zod schema strips unknown annotation fields, so # Anthropic-specific hints use _meta with namespaced keys instead. # maxResultSizeChars controls the CLI's layer-2 tool-result spill # threshold (toolResultStorage.ts maybePersistLargeToolResult). if tool_def.annotations is None: return None max_size = getattr(tool_def.annotations, "maxResultSizeChars", None) if max_size is None: return None return {"anthropic/maxResultSizeChars": max_size} cached_tool_list = [ Tool.model_validate( { "name": tool_def.name, "description": tool_def.description, "inputSchema": _build_schema(tool_def), "annotations": tool_def.annotations, "_meta": _build_meta(tool_def), } ) for tool_def in tools ] # Register list_tools handler to expose available tools @server.list_tools() # type: ignore[no-untyped-call,untyped-decorator] async def list_tools() -> list[Tool]: """Return the list of available tools.""" return cached_tool_list # Register call_tool handler to execute tools @server.call_tool() # type: ignore[untyped-decorator] async def call_tool(name: str, arguments: dict[str, Any]) -> Any: """Execute a tool by name with given arguments.""" if name not in tool_map: raise ValueError(f"Tool '{name}' not found") tool_def = tool_map[name] # Call the tool's handler with arguments result = await tool_def.handler(arguments) # Convert result to MCP format content: list[ TextContent | ImageContent | AudioContent | ResourceLink | EmbeddedResource ] = [] if "content" in result: for item in result["content"]: item_type = item.get("type") if item_type == "text": content.append(TextContent(type="text", text=item["text"])) elif item_type == "image": content.append( ImageContent( type="image", data=item["data"], mimeType=item["mimeType"], ) ) elif item_type == "resource_link": parts = [] link_name = item.get("name") uri = item.get("uri") desc = item.get("description") if link_name: parts.append(link_name) if uri: parts.append(str(uri)) if desc: parts.append(desc) content.append( TextContent( type="text", text="\n".join(parts) if parts else "Resource link", ) ) elif item_type == "resource": resource = item.get("resource") or {} if "text" in resource: content.append( TextContent(type="text", text=resource["text"]) ) else: logger.warning( "Binary embedded resource cannot be converted to text, skipping" ) else: logger.warning( "Unsupported content type %r in tool result, skipping", item_type, ) return CallToolResult( content=content, isError=result.get("is_error", False) ) # Return SDK server configuration return McpSdkServerConfig(type="sdk", name=name, instance=server) __all__ = [ # Main exports "query", "__version__", # Transport "Transport", "ClaudeSDKClient", # Types "PermissionMode", "EffortLevel", "McpServerConfig", "McpSdkServerConfig", "McpServerStatus", "McpServerStatusConfig", "McpServerConnectionStatus", "McpServerInfo", "McpStatusResponse", "McpToolAnnotations", "McpToolInfo", "UserMessage", "AssistantMessage", "SystemMessage", "TaskStartedMessage", "TaskProgressMessage", "TaskUpdatedMessage", "TaskNotificationMessage", "TaskNotificationStatus", "TaskUpdatedStatus", "TERMINAL_TASK_STATUSES", "TaskUsage", "ResultMessage", "DeferredToolUse", "RateLimitEvent", "RateLimitInfo", "RateLimitStatus", "RateLimitType", "StreamEvent", "Message", "ClaudeAgentOptions", "TaskBudget", "TextBlock", "ThinkingBlock", "ThinkingConfig", "ThinkingConfigAdaptive", "ThinkingConfigEnabled", "ThinkingConfigDisabled", "ToolUseBlock", "ToolResultBlock", "ServerToolName", "ServerToolUseBlock", "ServerToolResultBlock", "ContentBlock", "ContextUsageCategory", "ContextUsageResponse", # Tool callbacks "CanUseTool", "ToolPermissionContext", "PermissionResult", "PermissionResultAllow", "PermissionResultDeny", "PermissionUpdate", # Hook support "HookCallback", "HookContext", "HookInput", "HookEventMessage", "BaseHookInput", "PreToolUseHookInput", "PostToolUseHookInput", "PostToolUseFailureHookInput", "PostToolUseFailureHookSpecificOutput", "UserPromptSubmitHookInput", "StopHookInput", "SubagentStopHookInput", "PreCompactHookInput", "NotificationHookInput", "SubagentStartHookInput", "PermissionRequestHookInput", "NotificationHookSpecificOutput", "SubagentStartHookSpecificOutput", "PermissionRequestHookSpecificOutput", "HookJSONOutput", "HookMatcher", # Agent support "AgentDefinition", "SettingSource", # Plugin support "SdkPluginConfig", # Session listing "list_sessions", "get_session_info", "get_session_messages", "list_subagents", "get_subagent_messages", "SDKSessionInfo", "SessionMessage", # Session store "SessionKey", "SessionStore", "SessionStoreEntry", "SessionStoreFlushMode", "SessionStoreListEntry", "SessionSummaryEntry", "SessionListSubkeysKey", "InMemorySessionStore", "fold_session_summary", "MirrorErrorMessage", "project_key_for_directory", "import_session_to_store", # Session listing (SessionStore-backed async variants) "list_sessions_from_store", "get_session_info_from_store", "get_session_messages_from_store", "list_subagents_from_store", "get_subagent_messages_from_store", # Session mutations "rename_session", "tag_session", "delete_session", "fork_session", "ForkSessionResult", # Session mutations (SessionStore-backed async variants) "rename_session_via_store", "tag_session_via_store", "delete_session_via_store", "fork_session_via_store", # Beta support "SdkBeta", # Sandbox support "SandboxSettings", "SandboxNetworkConfig", "SandboxIgnoreViolations", # MCP Server Support "create_sdk_mcp_server", "tool", "SdkMcpTool", "ToolAnnotations", # Errors "ClaudeSDKError", "CLIConnectionError", "CLINotFoundError", "ProcessError", "CLIJSONDecodeError", ]