pydantic-ai-tool-system by existential-birds/beagle
npx skills add https://github.com/existential-birds/beagle --skill pydantic-ai-tool-system根据是否需要上下文,提供两种装饰器:
from pydantic_ai import Agent, RunContext
agent = Agent('openai:gpt-4o')
# @agent.tool - 第一个参数必须是 RunContext
@agent.tool
async def get_user_data(ctx: RunContext[MyDeps], user_id: int) -> str:
"""从数据库获取用户数据。
Args:
ctx: 包含依赖项的运行上下文。
user_id: 用户 ID。
"""
return await ctx.deps.db.get_user(user_id)
# @agent.tool_plain - 不允许包含上下文参数
@agent.tool_plain
def calculate_total(prices: list[float]) -> float:
"""计算总价。
Args:
prices: 需要求和的单价列表。
"""
return sum(prices)
RunContext[DepsType]RunContext 参数广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
Google 风格(默认):
@agent.tool_plain
async def search(query: str, limit: int = 10) -> list[str]:
"""搜索项目。
Args:
query: 搜索查询词。
limit: 返回的最大结果数。
"""
Sphinx 风格:
@agent.tool_plain(docstring_format='sphinx')
async def search(query: str) -> list[str]:
"""搜索项目。
:param query: 搜索查询词。
"""
工具可以返回多种类型:
# 字符串(直接返回)
@agent.tool_plain
def get_info() -> str:
return "Some information"
# Pydantic 模型(序列化为 JSON)
@agent.tool_plain
def get_user() -> User:
return User(name="John", age=30)
# 字典(序列化为 JSON)
@agent.tool_plain
def get_data() -> dict[str, Any]:
return {"key": "value"}
# ToolReturn 用于自定义内容类型
from pydantic_ai import ToolReturn, ImageUrl
@agent.tool_plain
def get_image() -> ToolReturn:
return ToolReturn(content=[ImageUrl(url="https://...")])
RunContext 提供:
@agent.tool
async def my_tool(ctx: RunContext[MyDeps]) -> str:
# 依赖项
db = ctx.deps.db
api = ctx.deps.api_client
# 模型信息
model_name = ctx.model.model_name
# 使用量跟踪
tokens_used = ctx.usage.total_tokens
# 重试信息
attempt = ctx.retry # 当前重试次数(从 0 开始)
max_retries = ctx.max_retries
# 消息历史
messages = ctx.messages
return "result"
根据每个请求动态修改工具:
from pydantic_ai.tools import ToolDefinition
async def prepare_tools(
ctx: RunContext[MyDeps],
tool_defs: list[ToolDefinition]
) -> list[ToolDefinition]:
"""根据上下文过滤或修改工具。"""
if ctx.deps.user_role != 'admin':
# 对非管理员隐藏管理工具
return [t for t in tool_defs if not t.name.startswith('admin_')]
return tool_defs
agent = Agent('openai:gpt-4o', prepare_tools=prepare_tools)
对工具进行分组和组合:
from pydantic_ai import FunctionToolset, CombinedToolset
# 创建工具集
db_tools = FunctionToolset()
@db_tools.tool
def query_users(name: str) -> list[dict]:
"""按名称查询用户。"""
...
@db_tools.tool
def update_user(id: int, data: dict) -> bool:
"""更新用户数据。"""
...
# 在 agent 中使用
agent = Agent('openai:gpt-4o', toolsets=[db_tools])
# 组合工具集
all_tools = CombinedToolset([db_tools, api_tools])
@agent.tool_plain
async def bad_tool(ctx: RunContext[MyDeps]) -> str: # 错误!
...
@agent.tool
def bad_tool(user_id: int) -> str: # 错误!
...
@agent.tool
def bad_tool(user_id: int, ctx: RunContext[MyDeps]) -> str: # 错误!
...
两者都可用,但异步更适用于 I/O 操作:
# 异步(适用于 I/O 操作)
@agent.tool
async def fetch_data(ctx: RunContext[Deps]) -> str:
return await ctx.deps.client.get('/data')
# 同步(适用于 CPU 密集型操作)
@agent.tool_plain
def compute(x: int, y: int) -> int:
return x * y
每周安装量
99
代码仓库
GitHub 星标数
40
首次出现
2026 年 1 月 20 日
安全审计
安装于
opencode80
gemini-cli80
codex79
github-copilot73
claude-code69
cursor68
Two decorators based on whether you need context:
from pydantic_ai import Agent, RunContext
agent = Agent('openai:gpt-4o')
# @agent.tool - First param MUST be RunContext
@agent.tool
async def get_user_data(ctx: RunContext[MyDeps], user_id: int) -> str:
"""Get user data from database.
Args:
ctx: The run context with dependencies.
user_id: The user's ID.
"""
return await ctx.deps.db.get_user(user_id)
# @agent.tool_plain - NO context parameter allowed
@agent.tool_plain
def calculate_total(prices: list[float]) -> float:
"""Calculate total price.
Args:
prices: List of prices to sum.
"""
return sum(prices)
RunContext[DepsType]RunContext parameterGoogle style (default):
@agent.tool_plain
async def search(query: str, limit: int = 10) -> list[str]:
"""Search for items.
Args:
query: The search query.
limit: Maximum results to return.
"""
Sphinx style:
@agent.tool_plain(docstring_format='sphinx')
async def search(query: str) -> list[str]:
"""Search for items.
:param query: The search query.
"""
Tools can return various types:
# String (direct)
@agent.tool_plain
def get_info() -> str:
return "Some information"
# Pydantic model (serialized to JSON)
@agent.tool_plain
def get_user() -> User:
return User(name="John", age=30)
# Dict (serialized to JSON)
@agent.tool_plain
def get_data() -> dict[str, Any]:
return {"key": "value"}
# ToolReturn for custom content types
from pydantic_ai import ToolReturn, ImageUrl
@agent.tool_plain
def get_image() -> ToolReturn:
return ToolReturn(content=[ImageUrl(url="https://...")])
RunContext provides:
@agent.tool
async def my_tool(ctx: RunContext[MyDeps]) -> str:
# Dependencies
db = ctx.deps.db
api = ctx.deps.api_client
# Model info
model_name = ctx.model.model_name
# Usage tracking
tokens_used = ctx.usage.total_tokens
# Retry info
attempt = ctx.retry # Current retry attempt (0-based)
max_retries = ctx.max_retries
# Message history
messages = ctx.messages
return "result"
Dynamically modify tools per-request:
from pydantic_ai.tools import ToolDefinition
async def prepare_tools(
ctx: RunContext[MyDeps],
tool_defs: list[ToolDefinition]
) -> list[ToolDefinition]:
"""Filter or modify tools based on context."""
if ctx.deps.user_role != 'admin':
# Hide admin tools from non-admins
return [t for t in tool_defs if not t.name.startswith('admin_')]
return tool_defs
agent = Agent('openai:gpt-4o', prepare_tools=prepare_tools)
Group and compose tools:
from pydantic_ai import FunctionToolset, CombinedToolset
# Create a toolset
db_tools = FunctionToolset()
@db_tools.tool
def query_users(name: str) -> list[dict]:
"""Query users by name."""
...
@db_tools.tool
def update_user(id: int, data: dict) -> bool:
"""Update user data."""
...
# Use in agent
agent = Agent('openai:gpt-4o', toolsets=[db_tools])
# Combine toolsets
all_tools = CombinedToolset([db_tools, api_tools])
@agent.tool_plain
async def bad_tool(ctx: RunContext[MyDeps]) -> str: # ERROR!
...
@agent.tool
def bad_tool(user_id: int) -> str: # ERROR!
...
@agent.tool
def bad_tool(user_id: int, ctx: RunContext[MyDeps]) -> str: # ERROR!
...
Both work, but async is preferred for I/O:
# Async (preferred for I/O operations)
@agent.tool
async def fetch_data(ctx: RunContext[Deps]) -> str:
return await ctx.deps.client.get('/data')
# Sync (fine for CPU-bound operations)
@agent.tool_plain
def compute(x: int, y: int) -> int:
return x * y
Weekly Installs
99
Repository
GitHub Stars
40
First Seen
Jan 20, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode80
gemini-cli80
codex79
github-copilot73
claude-code69
cursor68
AI Elements:基于shadcn/ui的AI原生应用组件库,快速构建对话界面
66,200 周安装