17.3 Skills 核心 API
核心 API 概述
Claude Code SDK 提供了丰富的 API 用于开发 Skills。本节将详细介绍 Skills 的核心 API，包括 Skill 基类、上下文管理、工具调用等。
python
## Skill 基类
### 1\. Skill 类
#### 1.1 基本结构
python
from typing import Dict, Any, Optional
from claude_code_sdk import Skill, SkillContext, SkillResult
class MySkill(Skill):
"""自定义 Skill 实现"""
def __init__(self):
super().__init__(
name="my-skill",
version="0.1.0",
description="A custom Claude Code skill"
)
def get_parameters_schema(self) -> Dict[str, Any]:
"""获取参数模式"""
return {
"type": "object",
"properties": {
"input": {
"type": "string",
"description": "Input text to process"
},
"options": {
"type": "object",
"description": "Additional options",
"properties": {
"format": {
"type": "string",
"enum": ["json", "yaml", "text"],
"default": "json"
},
"case": {
"type": "string",
"enum": ["upper", "lower", "title"],
"default": "upper"
}
}
}
},
"required": ["input"]
}
def execute(self, parameters: Dict[str, Any], context: SkillContext) -> SkillResult:
"""执行 Skill"""
try:
# 验证参数
self.validate_parameters(parameters)
# 处理输入
input_text = parameters["input"]
options = parameters.get("options", {})
# 处理文本
output = self.process_text(input_text, options)
# 返回结果
return SkillResult(
success=True,
data={
"input": input_text,
"output": output,
"options": options
}
)
except Exception as e:
return SkillResult(
success=False,
error=str(e)
)
def validate_parameters(self, parameters: Dict[str, Any]) -> bool:
"""验证参数"""
schema = self.get_parameters_schema()
required = schema.get("required", [])
for param in required:
if param not in parameters:
raise ValueError(f"Missing required parameter: {param}")
return True
def process_text(self, text: str, options: Dict[str, Any]) -> str:
"""处理文本"""
case = options.get("case", "upper")
if case == "upper":
return text.upper()
elif case == "lower":
return text.lower()
elif case == "title":
return text.title()
else:
return text
#### 1.2 Skill 属性
python
class MySkill(Skill):
def __init__(self):
super().__init__(
name="my-skill", # Skill 名称
version="0.1.0", # Skill 版本
description="Description", # Skill 描述
author="Author Name", # 作者
license="MIT", # 许可证
homepage="https://example.com" # 主页
)
# 自定义属性
self.custom_property = "value"
#### 1.3 Skill 方法
python
class MySkill(Skill):
def initialize(self):
"""初始化 Skill"""
# 加载配置
self.config = self.load_config()
# 初始化资源
self.resources = self.initialize_resources()
def load_config(self) -> Dict[str, Any]:
"""加载配置"""
# 从文件或环境变量加载配置
pass
def initialize_resources(self) -> Dict[str, Any]:
"""初始化资源"""
# 初始化数据库连接、缓存等
pass
def cleanup(self):
"""清理资源"""
# 关闭连接、释放资源
pass
def get_metadata(self) -> Dict[str, Any]:
"""获取元数据"""
return {
"name": self.name,
"version": self.version,
"description": self.description,
"author": self.author,
"capabilities": self.get_capabilities()
}
def get_capabilities(self) -> list:
"""获取能力列表"""
return [
"text_processing",
"format_conversion",
"case_conversion"
]
bash
### 2. SkillContext 类
#### 2.1 上下文属性
```python
from claude_code_sdk import SkillContext
```python
# SkillContext 提供以下属性：
# - project: 项目信息
# - user: 用户信息
# - environment: 环境信息
# - session: 会话信息
# - config: 配置信息
def use_context(context: SkillContext):
# 访问项目信息
project_name = context.project.name
project_path = context.project.path
# 访问用户信息
user_id = context.user.id
user_name = context.user.name
# 访问环境信息
environment = context.environment.name
platform = context.environment.platform
# 访问会话信息
session_id = context.session.id
start_time = context.session.start_time
# 访问配置信息
config = context.config.settings
```
#### 2.2 上下文方法
```python
def use_context_methods(context: SkillContext):
# 获取文件内容
file_content = context.read_file("src/main.py")
# 写入文件
context.write_file("output.txt", "Hello, World!")
# 搜索代码
results = context.search_codebase("function_name")
# 执行命令
output = context.run_command("ls -la")
# 获取环境变量
env_var = context.get_env_var("PATH")
# 记录日志
context.log("Info message", level="info")
context.log("Error message", level="error")
### 3\. SkillResult 类
#### 3.1 结果类型
python
from claude_code_sdk import SkillResult
# 成功结果
success_result = SkillResult(
success=True,
data={
"output": "processed text",
"metadata": {
"length": 14,
"words": 2
}
},
message="Processing completed successfully"
)
# 失败结果
error_result = SkillResult(
success=False,
error="Invalid input parameter",
error_code="INVALID_PARAMETER",
details={
"parameter": "input",
"expected": "string",
"received": "number"
}
)
# 部分成功结果
partial_result = SkillResult(
success=True,
data={
"processed": ["item1", "item2"],
"failed": ["item3"]
},
warnings=[
"item3 could not be processed"
]
)
#### 3.2 结果方法
python
def use_result_methods(result: SkillResult):
# 检查是否成功
if result.success:
print("Operation succeeded")
else:
print(f"Operation failed: {result.error}")
# 获取数据
data = result.data
# 获取错误信息
error = result.error
error_code = result.error_code
# 获取警告
warnings = result.warnings
# 获取消息
message = result.message
# 获取详细信息
details = result.details
# 转换为字典
result_dict = result.to_dict()
# 转换为 JSON
result_json = result.to_json()
bash
## 工具调用 API
### 1. 文件系统工具
#### 1.1 读取文件
```python
from claude_code_sdk import SkillContext
```python
def read_file_example(context: SkillContext):
# 读取文件内容
content = context.read_file("src/main.py")
# 读取部分内容
partial_content = context.read_file(
"src/main.py",
start_line=1,
end_line=10
)
# 读取二进制文件
binary_content = context.read_file_binary("image.png")
return content
```
#### 1.2 写入文件
```python
def write_file_example(context: SkillContext):
# 写入文本文件
context.write_file(
"output.txt",
"Hello, World!"
)
# 追加内容
context.write_file(
"output.txt",
"\nNew line",
mode="append"
)
# 写入二进制文件
binary_data = b'\x89PNG\r\n\x1a\n'
context.write_file_binary("image.png", binary_data)
#### 1.3 编辑文件
```python
def edit_file_example(context: SkillContext):
# 替换文本
context.edit_file(
"src/main.py",
old_text="old_function",
new_text="new_function"
)
# 使用正则表达式替换
context.edit_file_regex(
"src/main.py",
pattern=r"def (\w+)\(",
replacement=r"async def \1("
)
# 在指定位置插入
context.insert_file(
"src/main.py",
line_number=10,
text="# New comment"
)
#### 1.4 删除文件
```python
def delete_file_example(context: SkillContext):
# 删除文件
context.delete_file("output.txt")
# 删除目录
context.delete_directory("temp")
# 检查文件是否存在
exists = context.file_exists("src/main.py")
### 2\. 搜索工具
#### 2.1 代码搜索
python
def search_codebase_example(context: SkillContext):
# 搜索代码
results = context.search_codebase("function_name")
# 搜索特定文件类型
results = context.search_codebase(
"function_name",
file_pattern="*.py"
)
# 搜索特定目录
results = context.search_codebase(
"function_name",
directory="src"
)
# 使用正则表达式搜索
results = context.search_codebase_regex(
r"def \w+\(",
file_pattern="*.py"
)
return results
#### 2.2 文件搜索
python
def search_files_example(context: SkillContext):
# 查找文件
files = context.find_files("*.py")
# 查找特定目录
files = context.find_files(
"*.py",
directory="src"
)
# 递归查找
files = context.find_files(
"*.py",
recursive=True
)
return files
bash
### 3. 命令执行工具
#### 3.1 执行命令
```python
def run_command_example(context: SkillContext):
# 执行命令
```python
output = context.run_command("ls -la")
# 执行命令并捕获输出
output, error, return_code = context.run_command_with_output(
"python script.py"
)
# 在特定目录执行命令
output = context.run_command(
"ls -la",
working_directory="/tmp"
)
# 设置环境变量
output = context.run_command(
"echo $MY_VAR",
env_vars={"MY_VAR": "value"}
)
return output
```
#### 3.2 异步执行
```python
async def run_command_async_example(context: SkillContext):
# 异步执行命令
task = context.run_command_async("long_running_command")
# 等待完成
output = await task.wait()
# 检查状态
status = task.get_status()
# 取消任务
task.cancel()
### 4\. Git 工具
#### 4.1 Git 操作
python
def git_operations_example(context: SkillContext):
# 获取 Git 状态
status = context.git_status()
# 获取当前分支
branch = context.git_current_branch()
# 获取提交历史
commits = context.git_log(limit=10)
# 获取差异
diff = context.git_diff()
# 创建提交
context.git_commit("Commit message")
# 创建分支
context.git_create_branch("new-branch")
# 切换分支
context.git_checkout("new-branch")
return status
## 上下文管理 API
### 1\. 项目上下文
python
def project_context_example(context: SkillContext):
# 获取项目信息
project = context.project
# 项目名称
name = project.name
# 项目路径
path = project.path
# 项目类型
project_type = project.type
# 技术栈
tech_stack = project.tech_stack
# 配置
config = project.config
return {
"name": name,
"path": path,
"type": project_type,
"tech_stack": tech_stack,
"config": config
}
### 2. 用户上下文
```python
def user_context_example(context: SkillContext):
# 获取用户信息
```python
user = context.user
# 用户 ID
user_id = user.id
# 用户名
name = user.name
# 邮箱
email = user.email
# 角色
roles = user.roles
# 偏好
preferences = user.preferences
return {
"user_id": user_id,
"name": name,
"email": email,
"roles": roles,
"preferences": preferences
}
### 3. 环境上下文
def environment_context_example(context: SkillContext):
# 获取环境信息
environment = context.environment
# 环境名称
name = environment.name
# 平台
platform = environment.platform
# Python 版本
python_version = environment.python_version
# 工作目录
working_directory = environment.working_directory
# 环境变量
env_vars = environment.env_vars
return {
"name": name,
"platform": platform,
"python_version": python_version,
"working_directory": working_directory,
"env_vars": env_vars
}
~~~
## 事件处理 API
### 1. 事件监听
~~~
python
from claude_code_sdk import SkillContext
class MySkill(Skill):
def __init__(self):
super().__init__(
name="my-skill",
version="0.1.0",
description="A custom Claude Code skill"
)
def setup_event_listeners(self, context: SkillContext):
# 监听文件变更
context.on_file_changed(self.handle_file_changed)
# 监听命令执行
context.on_command_executed(self.handle_command_executed)
# 监听错误
context.on_error(self.handle_error)
def handle_file_changed(self, file_path: str):
"""处理文件变更事件"""
context.log(f"File changed: {file_path}")
def handle_command_executed(self, command: str, output: str):
"""处理命令执行事件"""
context.log(f"Command executed: {command}")
def handle_error(self, error: Exception):
"""处理错误事件"""
context.log(f"Error occurred: {error}", level="error")
### 2. 事件触发
def trigger_events_example(context: SkillContext):
# 触发自定义事件
context.emit_event(
"custom_event",
data={"key": "value"}
)
# 触发进度更新
context.emit_progress(
current=50,
total=100,
message="Processing..."
)
# 触发完成事件
context.emit_completion(
success=True,
message="Task completed"
)
~~~
## 缓存 API
### 1. 缓存操作
~~~
python
def cache_operations_example(context: SkillContext):
# 设置缓存
context.cache.set(
key="my_key",
value="my_value",
ttl=3600 # 1 小时
)
# 获取缓存
value = context.cache.get("my_key")
# 检查缓存是否存在
exists = context.cache.exists("my_key")
# 删除缓存
context.cache.delete("my_key")
# 清空缓存
context.cache.clear()
# 获取缓存统计
stats = context.cache.get_stats()
return value
### 2. 缓存策略
def cache_strategy_example(context: SkillContext):
# 使用 LRU 缓存
context.cache.set(
key="my_key",
value="my_value",
strategy="lru",
max_size=100
)
# 使用 TTL 缓存
context.cache.set(
key="my_key",
value="my_value",
strategy="ttl",
ttl=3600
)
# 使用永久缓存
context.cache.set(
key="my_key",
value="my_value",
strategy="permanent"
)
~~~
## 日志 API
### 1. 日志记录
~~~
python
def logging_example(context: SkillContext):
# 记录不同级别的日志
context.log("Debug message", level="debug")
context.log("Info message", level="info")
context.log("Warning message", level="warning")
context.log("Error message", level="error")
context.log("Critical message", level="critical")
# 使用格式化日志
context.log(
"Processing file: {file}",
level="info",
file="main.py"
)
# 记录结构化数据
context.log(
"Processing completed",
level="info",
data={
"files_processed": 10,
"time_elapsed": 5.2
}
)
### 2. 日志配置
```python
def logging_config_example(context: SkillContext):
# 配置日志级别
context.set_log_level("debug")
# 配置日志格式
context.set_log_format(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
# 配置日志输出
context.set_log_output(
file="/var/log/my-skill.log"
)