编写一个 Python 粒子
粒子是一个单文件程序,它向粒子运行时暴露一个或多个工具(命名操作,带 JSON Schema 输入)。 本技能告诉您生成一个可以构建和运行的粒子所需的一切。
1. 文件布局
项目根目录下的一个源文件,必须命名为 Particlefile.py。
不需要其他任何内容。PyPI 依赖项在文件顶部的 PEP 723 内联元数据块中声明;构建过程会解析它们。
无需 pyproject.toml,无需管理虚拟环境。
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "idna>=3", # 可选 — 仅当您使用第三方包时需要
# ]
# ///
2. 模块级别的 particle — 整个 API
文件必须将一个 Particle 实例赋值给模块级别的 particle 变量。
运行时在构建时内省该变量,并在每次工具调用时读取其处理器。
from particle.manifest import (
Particle, Tool, Http, Filesystem, Mount, TempMount,
Credential, OAuth2, ApiKey, ApiKeyLocation, SigningKey, Basic, Raw,
)
from particle import http, credentials, kv, oauth, signing
def _do_thing(args):
return {"result": f"hello {args['name']}"}
def _ping():
return {"status": "ok", "message": "alive"}
particle = Particle(
name="kebab-case-name", # 必需,kebab-case 格式
description="...", # 必需,一行
version="0.1.0", # 必需,语义化版本号
# 出站 HTTP 允许列表。未在此处列出的主机在 wire 边界被拒绝。
# 如果粒子不发出任何出站请求,则完全省略 `http=`。
http=Http(allowed_hosts=["api.example.com"]),
# 主机目录访问,默认关闭。请参阅下面的“文件系统”。
# filesystem=Filesystem(mounts={...}, temp={...}),
# 可选。按名称声明;用户在设置时选择一种方法。
credentials={
"example": Credential(
hosts=["api.example.com"],
required=True,
methods={
# 请参阅下面的“凭据方法”
},
),
},
# 必需。name → Tool(description, input_schema, handler)。
tools={
"do_thing": Tool(
description="该工具的功能 — 一行,为 LLM 编写。",
input_schema={
"type": "object",
"properties": {
"name": {"type": "string", "description": "..."},
},
"required": ["name"],
},
handler=_do_thing,
),
},
# 可选。由 `particle ping` 调用。
ping=_ping,
)
工具
description:为 LLM 调用者编写 — 简短、面向操作。input_schema:JSON Schema Draft 2020-12,根对象。主机在调用您的处理之前会根据它验证参数,因此您可以信任处理函数内部的字典形状。handler:一个普通的可调用对象,接受一个位置字典参数并返回任何 JSON 可序列化的值。可以是def或async def;异步处理器在一次性轮询循环上运行。抛出异常以呈现错误。
Ping
可选。返回一个类似这样的字典:
{"status": "ok" | "degraded" | "unhealthy",
"message": "...", # 可选
"details": ...} # 可选,JSON 可序列化
3. 处理器可用的运行时 API
所有主机能力都位于 from particle import ... 下。这些是您需要的唯一非标准库模块。
出站 HTTP
四种等效的 HTTP 调用方式。选择与您代码其余部分风格最匹配的一种;它们都通过主机允许列表进行路由。
particle.http.fetch — 原生 API。当设置了 credential_name= 时,自动应用凭据。
from particle import http
res = http.fetch(
"https://api.example.com/things",
method="POST",
headers={"Content-Type": "application/json"},
body=json.dumps({...}).encode(),
credential_name="example", # 可选;主机替换 secret
)
if not res.ok:
raise RuntimeError(f"上游 {res.status_code}")
data = res.json()
http.fetch 返回一个 Response,包含:
.status_code(整数),.ok(布尔值),.headers((名称, 字节)元组列表),.header(name)查找,.body(字节),.text(字符串),.json()(已解析)。
urllib.request.urlopen — 开箱即用。运行时对其进行猴子补丁,使其通过 wasi:http 进行路由。
import urllib.request
with urllib.request.urlopen("https://api.example.com/x") as r:
body = r.read()
urllib3 / requests — 开箱即用。运行时会注入一个自定义连接类,因此 requests.get(...) 等开箱即用。它们不会自动应用凭据 — 对于需要身份验证的请求,要么使用 http.fetch(..., credential_name=...),要么获取占位符并自行设置标头。
import requests
r = requests.get("https://api.example.com/x") # 有效
httpx — 开箱即用,同步和异步均可。导入时注入自定义 HTTPTransport / AsyncHTTPTransport。
import httpx
r = httpx.get("https://api.example.com/x")
原始套接字不可用。 socket.socket() 会抛出 OSError。
文件系统
默认关闭 — 处理器只能看到自己的包。声明 Filesystem 能力以获取主机目录访问权限,然后使用常规的 open() / pathlib 进行读写(运行时会将它们路由到 wasi:filesystem)。
from particle.manifest import Filesystem, Mount, TempMount
filesystem=Filesystem(
# *用户*映射到真实路径的主机目录 — 使用
# `particle mount <粒子名> <挂载点名称> <主机路径>` 持久化,或每次运行使用
# `--mount <挂载点名称>=<主机路径>`。
mounts={
"data": Mount(
description="报告写入的位置", # 显示在安装提示中
path="/mnt/data", # 沙箱内部的绝对路径
access="readwrite", # "readonly" | "readwrite"
required=True, # 映射前拒绝运行(默认为 False)
),
},
# 主机每次运行时新提供并在退出时清除的临时空间。
# 始终可读写;用户从不映射这些。
temp={
"work": TempMount(description="临时空间", path="/tmp/work", max_size="10MB"),
},
),
with open("/mnt/data/in.json") as f:
data = f.read()
with open("/mnt/data/out.json", "w") as f:
f.write(result)
- 处理器直接打开声明的
path;挂载点名称永远不出现在沙箱内部。 - 写入
readonly挂载点,或读取/写入任何声明挂载点之外的路径,都会抛出OSError。 - 如果用户从未映射
required挂载点,则在运行时失败,而非构建/安装时。 max_size是一个字节数,可选KB/MB/GB后缀(如"10MB");超出时写入失败。
凭据
from particle import credentials
# 用户在设置时选择的方法("oauth"、"pat" 等)或 None。
method = credentials.get_configured_method("example")
# "raw" 类型凭据的原始值。
value = credentials.get_raw("some-raw-cred")
# 低级:获取不透明的占位符 + 应用说明,
# 描述主机期望在何处替换它。大多数情况下
# 您希望使用 `http.fetch(..., credential_name=...)` 代替,
# 它在内部执行此操作。
info = credentials.get_placeholder("example")
键值存储
按粒子作用域。仅限字符串 — 如果需要,可将二进制数据 base64 编码。
在 Particle 上使用 kv=KV(enabled=True) 声明它 — 没有声明,每次调用都会引发 not-declared 错误(无需用户批准,只需声明):
from particle.manifest import Particle, KV
# particle = Particle(..., kv=KV(enabled=True))
from particle import kv
kv.set("cursor", "abc123")
v = kv.get("cursor") # str | None
kv.delete("cursor")
keys = kv.list("prefix:") # list[str] — 注意:会遮蔽内置函数
将模块导入为 kv(而不是 from particle.kv import list),这样 list 就不会遮蔽内置函数。
OAuth 刷新
仅当您拥有 OAuth2 凭据时。使用 credential_name= 调用 http.fetch 会自动刷新过期的令牌;仅当上游拒绝您认为仍然有效的令牌时才调用此函数。
from particle import oauth
oauth.refresh("example")
签名
仅适用于 SigningKey 凭据。密钥材料永远不会进入 Python。
from particle import signing
sig = signing.sign("webhook-key", payload_bytes)
ok = signing.verify("webhook-key", data_bytes, signature_bytes)
4. 凭据方法
每个 Credential 都有一个 methods={...} 映射。用户在设置时选择一种。数据类类型来自 particle.manifest。
OAuth2
"oauth": OAuth2(
description="通过 OAuth 登录",
flows=["authorization-code", "device-code"],
scopes=["read", "write"],
authorization_url="https://provider/oauth/authorize",
token_url="https://provider/oauth/token",
device_auth_url="https://provider/oauth/device/code", # 用于 "device-code"
),
有效的流字符串:"authorization-code"、"authorization-code-pkce"、"device-code"。
ApiKey
"pat": ApiKey(
description="使用个人访问令牌",
location=ApiKeyLocation(kind="auth-scheme", scheme="Bearer"),
),
ApiKeyLocation.kind 是以下之一:
"header"+name:<name>: <key>"auth-scheme"+scheme:Authorization: <scheme> <key>(例如Bearer)"query-param"+name:附加为?<name>=<key>
完全省略 location 以让用户在设置时提示。
Basic
"basic_auth": Basic(description="用户名 + 密码"),
替换为 Authorization: Basic <base64>。
SigningKey
"webhook": SigningKey(description="...", algorithm="hmac-sha256"),
有效值:"hmac-sha256"、"hmac-sha512"。通过 signing.sign 使用。
Raw
"api_key_blob": Raw(description="..."),
通过 credentials.get_raw(name) 使用。仅当您确实需要原始字节时才选择此类型 — 其他类型更安全,因为值永远不会进入 Python 内存。
5. PyPI 依赖
在文件顶部的 PEP 723 块中声明:
# /// script
# requires-python = ">=3.12"
# dependencies = [
# "httpx>=0.27",
# "pyjwt>=2",
# ]
# ///
约束:
- 仅纯 Python wheels。 任何需要 C 扩展(如 numpy、pydantic v1、lxml 等)的内容都无法加载。
- 标准库大部分可用 —
json、re、hashlib、hmac、base64、secrets、datetime、urllib.parse、asyncio、email等都可以工作。 ssl、原始socket、subprocess— 不可用。open()/pathlib仅适用于已声明文件系统挂载内的路径(参见上面的“文件系统”)以及粒子自己的/particle/包;任意主机路径都会被拒绝。
6. 关键规则
-
没有模块作用域的主机调用。 在文件顶部调用
http.fetch、kv.get、credentials.get_placeholder等操作,会在构建时失败(内省阶段会在陷阱存储下运行您的模块)。所有主机调用都必须在handler或ping函数内部进行。# ✗ 构建会失败 _cached_data = http.fetch("https://api.example.com/init") particle = Particle(...) # ✓ 在处理器内部 def _go(args): res = http.fetch("https://api.example.com/...") ... particle = Particle(tools={"go": Tool(..., handler=_go)}) -
声明每个您要请求的主机。 对
http=Http(allowed_hosts=[...])中缺少的主机的请求会被拒绝。错误会作为处理器内的异常出现。 -
Credential.hosts必须是 HTTP 允许列表的子集。 构建过程会拒绝绑定到粒子无法访问的主机的凭据。 -
name必须使用 kebab-case(小写,连字符)。注册表键是(name, version)。 -
version必须是有效的语义化版本。 例如0.1.0、1.2.3-rc.1。 -
工具处理器必须返回 JSON 可序列化的值。 普通的
dict、list、str、int、float、bool、None。数据类或datetime将无法序列化 — 在返回前进行转换。
7. 工作流程
在编写 Particlefile.py 之后:
particle build # 解析、解析依赖、内省、注册
particle ping <名称> # 验证 ping(如果已定义)
particle run <名称> # 列出工具
particle run <名称> <工具> --help # 显示工具的标志
particle run <名称> <工具> --foo=bar # 调用它
particle build 会在第一次交互式地引导用户完成凭据设置。如果构建失败,错误会指明阶段(import-scan、resolve-and-fetch、manifest-extract)和具体问题。
如果粒子声明了文件系统挂载,请在运行前映射它们(或在 particle run 上传递 --mount name=path 用于一次性运行):
particle mount <名称> <挂载点名称> <主机路径> # 保存持久映射
particle mount <名称> # 列出挂载点 + 当前映射
要将粒子作为独立可执行文件暴露(方便用于 Claude Code 技能和 shell 使用),particle link <名称> ./<名称> 会写入一个启动器,将其参数转发给 particle run <名称>。
8. 完整示例
一个暴露 GitHub 仓库查找工具的最小粒子:
# /// script
# requires-python = ">=3.12"
# dependencies = []
# ///
import json
from particle import http
from particle.manifest import Particle, Tool, Http, Credential, ApiKey, ApiKeyLocation
def _get_repo(args):
res = http.fetch(
f"https://api.github.com/repos/{args['owner']}/{args['repo']}",
headers={"Accept": "application/vnd.github+json"},
credential_name="github",
)
if not res.ok:
raise RuntimeError(f"GitHub {res.status_code}: {res.text}")
r = res.json()
return {
"full_name": r.get("full_name"),
"stars": r.get("stargazers_count"),
"language": r.get("language"),
"topics": r.get("topics", []),
}
particle = Particle(
name="github-repo-lookup",
description="Look up a GitHub repository by owner/repo",
version="0.1.0",
http=Http(allowed_hosts=["api.github.com"]),
credentials={
"github": Credential(
hosts=["api.github.com"],
required=True,
methods={
"pat": ApiKey(
description="GitHub personal access token",
location=ApiKeyLocation(kind="auth-scheme", scheme="Bearer"),
),
},
),
},
tools={
"get_repo": Tool(
description="Fetch metadata for a public GitHub repository",
input_schema={
"type": "object",
"properties": {
"owner": {"type": "string", "description": "Repository owner"},
"repo": {"type": "string", "description": "Repository name"},
},
"required": ["owner", "repo"],
},
handler=_get_repo,
),
},
)
构建并运行:
particle build # 交互式设置 GitHub PAT
particle run github-repo-lookup get_repo --owner=fastapi --repo=fastapi
预期输出:
{
"full_name": "fastapi/fastapi",
"stars": ...,
"language": "Python",
"topics": ["async", "fastapi", ...]
}