- 异步支持: 完全异步的 API，支持高并发操作
- 流式响应: 支持流式返回，实时获取思考过程中的回复
- 会話管理: 自动管理对话上下文和会话状态
- 进程池: 智能的进程池管理，自动复用和清理空闲进程
- 系统命令: 完整支持 Gemini CLI 的所有系统命令（/help, /clear, /stats 等）
- 文件引用: 支持 @ 语法引用文件内容
- Shell 集成: 支持 ! 语法执行 shell 命令
- 配置灵活: 支持文件配置和环境变量配置
- 错误处理: 完善的异常处理和错误恢复机制
- 类型安全: 完整的类型注解支持
- 易于使用: 简洁的 API 设计，支持一行代码调用
从源码安装：
git clone https://github.com/example/gemini-cli-sdk.git
cd gemini-cli-sdk
pip install -e .
确保已安装 Gemini CLI 工具并且在 PATH 中可用：
# 检查 Gemini CLI 是否可用
gemini --version
import asyncio
from gemini_cli_sdk import GeminiClient
async def main():
# 使用上下文管理器自动管理资源
async with GeminiClient() as client:
# 简单的一次性查询
response = await client.one_shot("什么是 Python？")
print(response.content)
asyncio.run(main())
import asyncio
from gemini_cli_sdk import GeminiClient
async def main():
async with GeminiClient() as client:
# 创建会话以维护对话上下文
session_id = client.create_session()
# 在会话中进行对话
response1 = await client.chat("你好，我正在学习编程", session_id)
response2 = await client.chat("能给我一些建议吗？", session_id)
print(f"AI: {response1.content}")
print(f"AI: {response2.content}")
# 清理会话
client.close_session(session_id)
asyncio.run(main())
import asyncio
import json # 导入json模块
from gemini_cli_sdk import GeminiClient
async def main():
async with GeminiClient() as client:
print("AI (流式): ", end="")
async for chunk in await client.chat("写一首关于宇宙的短诗", stream=True):
try:
json_chunk = json.loads(chunk)
if json_chunk.get("type") == "message" and json_chunk.get("role") == "assistant":
content_part = json_chunk.get("content", "")
print(content_part, end="", flush=True)
except json.JSONDecodeError:
# 处理非JSON格式的块，或者错误情况
print(chunk, end="", flush=True) # 作为回退，直接打印原始内容
print() # 流式输出结束后换行
asyncio.run(main())
import asyncio
from gemini_cli_sdk import GeminiClient
async def main():
async with GeminiClient() as client:
# 顺序发送多个消息（共享上下文）
messages = [
"介绍一下机器学习",
"深度学习和机器学习有什么区别？",
"推荐一些学习资源"
]
session_id = client.create_session()
responses = await client.send_batch(messages, session_id)
for i, response in enumerate(responses):
print(f"Q{i+1}: {messages[i]}")
print(f"A{i+1}: {response.content}\n")
asyncio.run(main())
import asyncio
from gemini_cli_sdk import GeminiClient
async def main():
async with GeminiClient() as client:
# 并发发送多个独立的消息
questions = [
"什么是 Python？",
"什么是 JavaScript？",
"什么是 Go？"
]
responses = await client.send_concurrent(questions)
for question, response in zip(questions, responses):
print(f"Q: {question}")
print(f"A: {response.content}\n")
asyncio.run(main())
创建 ~/.gemini_cli_sdk.json 配置文件：
{
"max_processes": 5,
"idle_timeout": 300,
"max_context_length": 50,
"gemini_command": "gemini",
"gemini_args": [],
"enable_logging": true,
"log_level": "INFO",
"response_timeout": 30.0,
"cleanup_interval": 60
}
export GEMINI_MAX_PROCESSES=5
export GEMINI_IDLE_TIMEOUT=300
export GEMINI_MAX_CONTEXT_LENGTH=50
export GEMINI_COMMAND=gemini
export GEMINI_ARGS=""
export GEMINI_ENABLE_LOGGING=true
export GEMINI_LOG_LEVEL=INFO
export GEMINI_RESPONSE_TIMEOUT=30.0
export GEMINI_CLEANUP_INTERVAL=60
from gemini_cli_sdk import GeminiClient, GeminiConfig
config = GeminiConfig(
max_processes=3,
idle_timeout=600,
max_context_length=100,
response_timeout=60.0
)
async with GeminiClient(config=config) as client:
response = await client.chat("Hello!")
print(response.content)
主要的客户端类，提供与 Gemini CLI 交互的接口。
async start(): 启动客户端async stop(): 停止客户端create_session(user_id=None, metadata=None): 创建新会话close_session(session_id): 关闭会话async send_message(message, session_id=None, system_instruction=None, maintain_context=True, stream=False): 发送消息，根据stream参数返回GeminiResponse或异步生成器async chat(message, session_id=None, stream=False): 简单聊天接口，支持流式返回async one_shot(message, system_instruction=None, stream=False): 一次性查询，支持流式返回async send_batch(messages, session_id=None, system_instruction=None): 批量发送async send_concurrent(messages, system_instruction=None): 并发发送get_stats(): 获取客户端统计信息async health_check(): 健康检查
@dataclass
class GeminiResponse:
content: str
session_id: Optional[str] = None
process_id: Optional[str] = None
timestamp: float = field(default_factory=time.time)
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class SessionInfo:
session_id: str
created_at: float
last_activity: float
message_count: int
process_id: Optional[str] = None
metadata: Dict[str, Any] = field(default_factory=dict)
GeminiSDKError: 基础异常类GeminiProcessError: 进程相关错误GeminiSessionError: 会话相关错误GeminiConfigError: 配置相关错误GeminiTimeoutError: 超时错误GeminiNotFoundError: Gemini CLI 未找到GeminiConnectionError: 连接错误GeminiValidationError: 输入验证错误
async with GeminiClient() as client:
system_instruction = "你是一个专业的 Python 编程助手，请用简洁明了的方式回答问题。"
response = await client.one_shot(
"如何创建一个 Python 类？",
system_instruction=system_instruction
)
print(response.content)
async with GeminiClient() as client:
# 创建带元数据的会话
session_id = client.create_session(
user_id="user123",
metadata={"topic": "programming", "level": "beginner"}
)
# 获取会话信息
session_info = client.get_session_info(session_id)
print(f"会话创建时间: {session_info.created_at}")
print(f"消息数量: {session_info.message_count}")
async with GeminiClient() as client:
# 获取客户端统计信息
stats = client.get_stats()
print(f"活跃会话数: {stats['sessions']['active']}")
print(f"进程池状态: {stats['processes']['idle']}/{stats['processes']['total']}")
# 健康检查
health = await client.health_check()
print(f"健康状态: {health['status']}")
SDK 完整支持 Gemini CLI 的所有系统命令：
async with GeminiClient() as client:
# 获取帮助信息
help_result = await client.system_help()
print(help_result['available_commands'])
# 获取版本信息
about_result = await client.system_about()
print(about_result['version'])
# 获取系统统计
stats_result = await client.system_stats('model')
print(stats_result)
# 清除会话历史
await client.system_clear(session_id)
# 管理对话检查点
await client.chat_save_checkpoint("my_checkpoint", session_id)
checkpoints = await client.chat_list_checkpoints()
await client.chat_resume_checkpoint("my_checkpoint")
# 管理扩展
extensions = await client.manage_extensions("list")
await client.manage_extensions("update", "extension_name")
# 管理 MCP 服务器
mcp_servers = await client.manage_mcp_servers("list")
await client.manage_mcp_servers("refresh")
# 管理工作空间目录
await client.manage_workspace_directories("add", "/path/to/dir")
dirs = await client.manage_workspace_directories("show")
# 管理内存
await client.manage_memory("add", "重要信息")
memory = await client.manage_memory("show")
# 列出可用工具
tools = await client.list_tools(show_descriptions=True)
支持文件引用、Shell 命令和系统命令的统一接口：
async with GeminiClient() as client:
# 使用系统命令
response = await client.send_message_with_features("/help")
print(response.metadata['command_result'])
# 使用文件引用（@ 语法）
response = await client.send_message_with_features(
"分析这个文件 @src/main.py 的代码结构",
process_file_refs=True
)
# 使用 Shell 命令（! 语法）
response = await client.send_message_with_features(
"!ls -la",
allow_shell_commands=True
)
# 直接执行系统命令
result = await client.execute_system_command("stats", ["model"])
print(result)
pip install -e ".[dev]"
# 运行所有测试
pytest
# 运行详细测试输出
pytest -v
# 运行特定测试文件
pytest tests/test_basic.py -v
pytest tests/test_system_commands.py -v
# 运行测试并显示覆盖率
pytest --cov=gemini_cli_sdk tests/
# 基础使用示例
PYTHONPATH=. python3 examples/basic_usage.py
# 配置示例
PYTHONPATH=. python3 examples/configuration_example.py
# 系统命令示例
PYTHONPATH=. python3 examples/system_commands_example.py
# 高级功能示例（文件引用、Shell命令等）
PYTHONPATH=. python3 examples/advanced_features_example.py
如果你想测试文件引用和Shell命令功能，可以运行高级功能示例：
cd gemini-cli-sdk
PYTHONPATH=. python3 examples/advanced_features_example.py
这个示例会演示：
- ✅ 文件引用处理 (
@filename语法) - ✅ Shell命令执行 (
!command语法) - ✅ 系统命令集成 (
/command语法) - ✅ 统一消息接口
- ✅ 安全特性（危险命令阻止、文件大小限制等）
SDK 默认启用日志功能，你可以看到详细的操作日志：
# 运行示例时会看到日志输出
PYTHONPATH=. python3 examples/advanced_features_example.py
# 运行测试时查看日志
PYTHONPATH=. python3 -m pytest tests/test_logging.py -v -s
# 测试特定功能的日志
PYTHONPATH=. python3 -m pytest tests/test_system_commands.py -v -s
日志输出示例：
2025-12-27 13:48:08,851 - gemini_cli_sdk.process_manager - INFO - Verified Gemini CLI at: gemini
2025-12-27 13:48:08,851 - gemini_cli_sdk.process_manager - INFO - Process manager started
2025-12-27 13:48:08,851 - gemini_cli_sdk.client - INFO - Gemini client started
2025-12-27 13:48:08,851 - gemini_cli_sdk.system_commands - INFO - Executed command: /help
2025-12-27 13:48:08,870 - gemini_cli_sdk.client - INFO - Executed shell command: echo Hello World (return code: 0)
你可以通过配置文件或环境变量调整日志级别：
# 设置为 DEBUG 级别查看更详细的日志
export GEMINI_LOG_LEVEL=DEBUG
PYTHONPATH=. python3 examples/advanced_features_example.py
# 或者禁用日志
export GEMINI_ENABLE_LOGGING=false
PYTHONPATH=. python3 examples/advanced_features_example.py
black gemini_cli_sdk/
mypy gemini_cli_sdk/
MIT License
欢迎提交 Issue 和 Pull Request！
- 初始版本发布
- 支持基本的 Gemini CLI 交互
- 会话管理功能
- 进程池管理
- 配置系统
- 完整的异常处理