pi-mono自定义工具开发从零开始：基础入门到实战指南

1 认识pi-mono工具生态

在AI助手领域，工具就像是智能体的"双手"，让AI从被动响应转变为主动解决问题的实体。pi-mono作为一款功能完备的AI agent工具包，不仅提供丰富的内置能力，更通过开放的扩展系统让用户能够根据自身需求打造专属工具。

工具生态的核心价值

想象一下，标准AI助手就像一部基础手机，而pi-mono的工具扩展系统则让你可以安装各种"应用程序"，将基础手机变成专业设备。这种扩展能力让pi-mono能够适应从软件开发到数据分析的各种专业场景。

pi-mono的工具系统具有三大特性：

• 模块化设计：每个工具独立封装，便于开发和维护
• 自动发现机制：系统能自动识别并加载符合规范的工具
• 统一调用接口：所有工具遵循相同的调用模式，降低使用门槛

图1：pi-mono交互式模式界面，展示了工具和扩展的使用环境，alt文本：pi-mono工具集成与API集成的交互式操作界面

工具类型与应用场景

pi-mono工具生态包含多种类型：

• 系统工具：文件操作、命令执行等基础功能
• 数据处理工具：文本分析、格式转换等数据处理能力
• API对接工具：连接外部服务的桥梁
• 专业领域工具：针对特定行业的专业功能

这些工具共同构成了pi-mono的能力扩展体系，让AI助手能够处理复杂的实际问题。

知识点卡片：pi-mono工具生态的核心价值在于其开放性和模块化设计，允许用户根据需求扩展AI能力边界。工具系统采用自动发现机制，降低了集成门槛，同时统一的调用接口确保了使用体验的一致性。

2 自定义工具开发基础

开发自定义工具是扩展pi-mono能力的关键方式。本章节将介绍工具开发的基础知识，包括环境准备、项目结构和基础模板。

开发环境搭建

在开始开发前，需要准备以下环境：

1. Python环境：推荐Python 3.8及以上版本

   # 检查Python版本
   python --version
   

2. pi-mono开发环境：

   # 克隆项目仓库
   git clone https://gitcode.com/GitHub_Trending/pi/pi-mono
   cd pi-mono
   
   # 安装依赖
   npm install
   

3. 工具开发目录：

   # 创建工具开发目录
   mkdir -p ~/.pi/agent/tools/my_first_tool
   cd ~/.pi/agent/tools/my_first_tool
   

工具文件结构

pi-mono采用约定优于配置的原则，工具需要遵循特定的文件结构：

~/.pi/agent/tools/
  my_first_tool/          # 工具目录
    __init__.py           # 包初始化文件
    main.py               # 工具入口文件
    requirements.txt      # 依赖声明文件
    README.md             # 工具说明文档

这种结构就像一个标准化的"产品包装"，让pi-mono能够快速识别和加载你的工具。

Python工具基础模板

以下是一个Python工具的基础模板：

# ~/.pi/agent/tools/my_first_tool/main.py
from pi_toolkit import Tool, ToolContext, Parameter

def create_my_tool() -> Tool:
    """创建自定义工具"""
    return Tool(
        name="greeting_tool",
        description="生成个性化问候语的工具",
        parameters=[
            Parameter(
                name="name",
                type="string",
                description="用户姓名",
                required=True
            ),
            Parameter(
                name="occasion",
                type="string",
                description="问候场合，如'生日'、'节日'或'日常'",
                required=False,
                default="日常"
            )
        ],
        execute=execute_greeting
    )

async def execute_greeting(ctx: ToolContext, params: dict) -> str:
    """执行工具逻辑"""
    name = params["name"]
    occasion = params.get("occasion", "日常")
    
    greetings = {
        "日常": f"你好，{name}！很高兴见到你。",
        "生日": f"生日快乐，{name}！愿你今天充满喜悦。",
        "节日": f"节日快乐，{name}！祝你度过愉快的时光。"
    }
    
    return greetings.get(occasion, greetings["日常"])

这个模板定义了一个简单的问候语生成工具，包含工具元数据和执行逻辑两部分。

知识点卡片：Python工具开发需要遵循特定的文件结构和模板格式。工具定义包含名称、描述、参数规范和执行函数四个核心部分。pi-mono会自动发现符合规范的工具并将其集成到系统中，无需额外配置。

3 构建金融数据查询工具

现在我们将通过一个实战案例，开发一个对接第三方金融数据API的工具，展示pi-mono工具开发的完整流程。

需求分析与设计

问题：投资者需要快速获取股票市场数据，但手动查询多个来源效率低下。

方案：开发一个金融数据查询工具，集成第三方股票API，提供实时行情查询功能。

工具功能设计：

• 查询股票实时价格
• 获取股票历史数据
• 显示公司基本信息
• 支持多市场股票代码

工具实现步骤

1. 创建工具目录结构

mkdir -p ~/.pi/agent/tools/financial_data
cd ~/.pi/agent/tools/financial_data
touch __init__.py main.py requirements.txt README.md

2. 安装依赖

# requirements.txt
requests>=2.25.1
python-dotenv>=0.19.0

安装依赖：

pip install -r requirements.txt

3. 实现工具代码

# main.py
import os
import requests
from dotenv import load_dotenv
from pi_toolkit import Tool, ToolContext, Parameter

# 加载环境变量
load_dotenv()

def create_financial_tool() -> Tool:
    """创建金融数据查询工具"""
    return Tool(
        name="financial_data",
        description="查询股票市场数据的工具，支持实时价格和历史数据",
        parameters=[
            Parameter(
                name="symbol",
                type="string",
                description="股票代码，如'AAPL'或'000001.SS'",
                required=True
            ),
            Parameter(
                name="data_type",
                type="string",
                description="数据类型：'price'表示实时价格，'history'表示历史数据，'info'表示公司信息",
                required=True,
                enum=["price", "history", "info"]
            ),
            Parameter(
                name="days",
                type="integer",
                description="查询历史数据的天数，仅当data_type为'history'时有效",
                required=False,
                default=30
            )
        ],
        execute=execute_financial_query
    )

async def execute_financial_query(ctx: ToolContext, params: dict) -> str:
    """执行金融数据查询"""
    # 获取API密钥
    api_key = os.getenv("FINANCIAL_API_KEY")
    if not api_key:
        return "错误：未配置FINANCIAL_API_KEY环境变量"
    
    symbol = params["symbol"]
    data_type = params["data_type"]
    
    # 构建API请求
    base_url = "https://financial-api.example.com/v1"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    try:
        if data_type == "price":
            # 查询实时价格
            response = requests.get(f"{base_url}/quotes/{symbol}", headers=headers)
            data = response.json()
            
            if "error" in data:
                return f"查询失败：{data['error']}"
                
            return (f"股票 {data['symbol']} ({data['name']}) 实时数据：\n"
                    f"价格：{data['latestPrice']} {data['currency']}\n"
                    f"变动：{data['change']} ({data['changePercent']}%)\n"
                    f"更新时间：{data['latestUpdate']}")
                    
        elif data_type == "history":
            # 查询历史数据
            days = params.get("days", 30)
            response = requests.get(
                f"{base_url}/historical/{symbol}?days={days}", 
                headers=headers
            )
            data = response.json()
            
            if "error" in data:
                return f"查询失败：{data['error']}"
                
            # 格式化历史数据
            result = f"股票 {data['symbol']} 过去{days}天历史数据：\n"
            result += "日期       收盘价    变动%\n"
            result += "------------------------\n"
            
            # 只显示前5条和后5条数据
            for entry in data['historical'][:5] + data['historical'][-5:]:
                result += f"{entry['date']}  {entry['close']:8.2f}  {entry['changePercent']:6.2f}%\n"
                
            return result
            
        elif data_type == "info":
            # 查询公司信息
            response = requests.get(f"{base_url}/company/{symbol}", headers=headers)
            data = response.json()
            
            if "error" in data:
                return f"查询失败：{data['error']}"
                
            return (f"公司信息：{data['name']} ({data['symbol']})\n"
                    f"行业：{data['sector']} / {data['industry']}\n"
                    f"总部：{data['address']}\n"
                    f"员工：{data['employees']}人\n"
                    f"市值：{data['marketCap']} {data['currency']}\n"
                    f"简介：{data['description'][:200]}...")
                    
    except Exception as e:
        return f"API请求错误：{str(e)}"

4. 工具注册与测试

创建工具入口文件：

# __init__.py
from .main import create_financial_tool

def load_tool():
    return create_financial_tool()

现在可以通过pi-mono的交互式模式测试工具：

pi interactive

在交互界面中，尝试调用工具：

使用financial_data工具查询AAPL的实时价格

工具配置与部署

为了安全管理API密钥，建议使用环境变量或pi-mono的设置系统：

1. 使用环境变量：

   export FINANCIAL_API_KEY="your_api_key_here"
   

2. 使用pi-mono设置文件（推荐）：

   // ~/.pi/agent/settings.json
   {
     "apiKeys": {
       "financial_api": "your_api_key_here"
     }
   }
   

   然后在工具中使用：

   api_key = await ctx.settings.get_api_key("financial_api")
   

知识点卡片：实战工具开发需要考虑功能设计、错误处理、安全配置等多方面因素。金融数据工具展示了如何对接第三方API、处理不同类型的请求参数、以及安全管理API密钥。工具的参数设计应清晰明确，错误处理需友好且具有诊断价值。

4 工具高级开发技巧

掌握基础工具开发后，我们来探索一些高级技巧，提升工具的质量和性能。

异步工具开发

在处理网络请求或耗时操作时，异步编程能显著提升工具性能：

# 异步HTTP请求示例
import aiohttp

async def execute_async_query(ctx: ToolContext, params: dict) -> str:
    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as response:
            data = await response.json()
            # 处理数据...

异步工具的优势：

• 非阻塞执行，不影响其他工具和UI响应
• 更高效地处理多个并发请求
• 适合I/O密集型任务

工具版本兼容性

随着pi-mono的版本更新，工具可能需要适配不同版本的API：

from pi_toolkit import __version__ as toolkit_version
from packaging import version

def create_compatible_tool() -> Tool:
    # 检查工具包版本
    if version.parse(toolkit_version) >= version.parse("1.0.0"):
        # 使用新版本API
        return Tool(
            name="compatible_tool",
            description="兼容新老版本的工具",
            parameters_v2=[...],  # 新版本参数格式
            execute=execute_new
        )
    else:
        # 兼容旧版本API
        return Tool(
            name="compatible_tool",
            description="兼容新老版本的工具",
            parameters=[...],  # 旧版本参数格式
            execute=execute_old
        )

工具单元测试

为确保工具质量，编写单元测试是最佳实践：

# test_financial_tool.py
import pytest
from unittest.mock import patch, MagicMock
from main import execute_financial_query

@pytest.mark.asyncio
async def test_price_query():
    # 创建模拟上下文
    ctx = MagicMock()
    
    # 模拟API响应
    with patch('main.requests.get') as mock_get:
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "symbol": "AAPL",
            "name": "Apple Inc",
            "latestPrice": 150.25,
            "currency": "USD",
            "change": 2.5,
            "changePercent": 1.7,
            "latestUpdate": "2023-06-15T16:00:00Z"
        }
        mock_get.return_value = mock_response
        
        # 执行测试
        result = await execute_financial_query(
            ctx, 
            {"symbol": "AAPL", "data_type": "price"}
        )
        
        # 验证结果
        assert "Apple Inc" in result
        assert "150.25 USD" in result

性能优化策略

1. 结果缓存：减少重复API请求

   async def execute_with_cache(ctx: ToolContext, params: dict) -> str:
       cache_key = f"financial:{params['symbol']}:{params['data_type']}"
       
       # 尝试从缓存获取
       cached_result = await ctx.cache.get(cache_key)
       if cached_result:
           return cached_result
           
       # 执行实际查询
       result = await execute_financial_query(ctx, params)
       
       # 缓存结果（设置10分钟过期）
       await ctx.cache.set(cache_key, result, ttl=600)
       
       return result
   

2. 请求批处理：合并多个请求减少API调用

   # 批量查询多个股票
   async def batch_query(symbols):
       # 将多个单股票查询合并为一个批量查询
       if len(symbols) > 1:
           return await fetch_batch_data(symbols)
       return await fetch_single_data(symbols[0])
   

工具事件通信

工具可以通过事件系统与其他组件通信：

async def execute_with_events(ctx: ToolContext, params: dict) -> str:
    # 发送开始事件
    ctx.events.emit("financial_query_started", {
        "symbol": params["symbol"],
        "data_type": params["data_type"]
    })
    
    try:
        result = await execute_financial_query(ctx, params)
        
        # 发送成功事件
        ctx.events.emit("financial_query_completed", {
            "symbol": params["symbol"],
            "data_type": params["data_type"],
            "success": True
        })
        
        return result
    except Exception as e:
        # 发送失败事件
        ctx.events.emit("financial_query_completed", {
            "symbol": params["symbol"],
            "data_type": params["data_type"],
            "success": False,
            "error": str(e)
        })
        raise

图2：pi-mono会话树视图，展示了工具调用历史和上下文切换，alt文本：pi-mono工具调用历史与API集成流程展示

知识点卡片：高级工具开发涉及异步编程、版本兼容、单元测试和性能优化等方面。异步工具能提升并发处理能力，版本兼容确保工具在不同环境下正常工作，单元测试保障代码质量，而缓存和批处理等技巧可以显著提升工具性能。

5 常见问题诊断与解决方案

在工具开发和使用过程中，可能会遇到各种问题。本章节总结了常见问题及解决方法。

工具加载失败

问题：工具未出现在pi-mono的可用工具列表中。

解决方案：

1. 检查工具目录结构是否符合规范

   # 正确结构示例
   ~/.pi/agent/tools/
     my_tool/
       __init__.py
       main.py
   

2. 验证工具入口函数是否正确

   # __init__.py中必须有load_tool函数
   def load_tool():
       return create_my_tool()
   

3. 查看pi-mono日志获取详细错误信息

   pi logs --tools
   

API调用错误

问题：工具调用第三方API时失败。

解决方案：

1. 检查API密钥是否正确配置

   # 正确获取API密钥的方式
   api_key = await ctx.settings.get_api_key("service_name")
   

2. 验证网络连接和代理设置

   # 检查代理配置
   proxies = ctx.settings.get("network.proxies", {})
   

3. 添加详细的错误处理和日志

   try:
       response = requests.get(url)
       response.raise_for_status()  # 抛出HTTP错误
   except requests.exceptions.HTTPError as e:
       ctx.logger.error(f"API请求失败: {str(e)}")
       return f"请求错误: {str(e)}"
   except requests.exceptions.ConnectionError:
       ctx.logger.error("网络连接失败")
       return "网络错误: 无法连接到API服务器"
   

性能问题

问题：工具执行缓慢或占用过多资源。

解决方案：

1. 实现结果缓存机制
2. 优化API调用次数，使用批处理
3. 采用异步I/O操作
4. 对大型结果进行分页或截断处理

# 大型结果截断示例
def truncate_large_result(result: str, max_length: int = 2000) -> str:
    if len(result) <= max_length:
        return result
        
    return (f"{result[:max_length]}...\n"
            f"结果已截断，完整结果长度: {len(result)}字符")

参数验证失败

问题：工具调用时提示参数错误。

解决方案：

1. 确保参数定义与实际使用一致
2. 提供清晰的参数描述
3. 添加参数验证逻辑

# 参数验证示例
async def execute_with_validation(ctx: ToolContext, params: dict) -> str:
    # 验证必填参数
    if "symbol" not in params:
        return "错误：缺少必填参数'symbol'"
        
    # 验证参数格式
    if not re.match(r'^[A-Z0-9]+\.?[A-Z]{2,3}$', params["symbol"]):
        return "错误：股票代码格式无效，应为如'AAPL'或'000001.SS'的格式"
        
    # 执行实际逻辑
    return await execute_financial_query(ctx, params)

知识点卡片：工具开发中常见问题包括加载失败、API调用错误、性能问题和参数验证失败等。通过规范目录结构、正确配置API密钥、实现缓存机制和添加参数验证等方法，可以有效解决这些问题。详细的错误日志和完善的异常处理是诊断问题的关键。

6 工具开发最佳实践总结

经过前面的学习，我们已经掌握了pi-mono自定义工具开发的核心知识和实战技巧。本章节将总结工具开发的最佳实践，帮助你构建高质量的pi-mono工具。

工具设计原则

1. 单一职责：每个工具应专注于解决特定问题，避免功能过于复杂
2. 接口清晰：参数定义明确，返回结果格式一致
3. 错误友好：提供清晰的错误信息和恢复建议
4. 文档完善：包含使用说明、参数解释和示例

代码质量保障

1. 单元测试：为工具编写全面的单元测试
2. 类型提示：使用Python类型提示提高代码可读性和稳定性
3. 代码规范：遵循PEP 8规范，保持代码风格一致
4. 版本控制：对工具代码进行版本管理

安全最佳实践

1. 密钥管理：使用pi-mono的API密钥管理系统，避免硬编码密钥
2. 输入验证：严格验证所有用户输入，防止注入攻击
3. 权限控制：根据需要限制工具的系统访问权限
4. 数据清理：对外部API返回的数据进行安全处理

性能优化 checklist

• [ ] 实现结果缓存机制
• [ ] 使用异步I/O处理网络请求
• [ ] 优化API调用次数
• [ ] 对大型结果进行分页或截断
• [ ] 避免在工具中执行耗时操作

工具分发与共享

当你开发了高质量的工具后，可以通过以下方式分享给其他pi-mono用户：

1. 本地工具包：打包为tar.gz文件分享

   cd ~/.pi/agent/tools
   tar -czf my_tool.tar.gz my_tool/
   

2. npm包分发：将工具发布为npm包

   // package.json
   {
     "name": "pi-financial-tool",
     "version": "1.0.0",
     "pi": {
       "type": "tool",
       "main": "dist/main.js"
     }
   }
   

3. 社区贡献：提交PR到pi-mono官方工具库

知识点卡片：高质量工具开发需要遵循单一职责、接口清晰、错误友好和文档完善等原则。通过单元测试、类型提示和代码规范保障代码质量，采用安全的密钥管理和输入验证保护系统安全。性能优化和良好的分发机制能让工具发挥更大价值。

通过本文的学习，你已经具备了开发pi-mono自定义工具的全面知识。无论是简单的辅助工具还是复杂的API集成工具，这些知识都将帮助你扩展pi-mono的能力边界，打造更强大的AI助手体验。