boykopovar/Grok3API

GitHub: boykopovar/Grok3API

一个基于逆向工程的异步 Python 库,通过移动端 gRPC 协议与 Grok 交互,提供流式响应、对话管理和完整的工具调用支持。

Stars: 56 | Forks: 8

# Grok3API ![Python](https://img.shields.io/badge/python-3.8%2B-blue?logo=python\&logoColor=white) ![Reverse engineered](https://img.shields.io/badge/Reverse%20engineered%20with-🔬-red) ![Stars](https://img.shields.io/github/stars/boykopovar/Grok3API?style=social) ![Forks](https://img.shields.io/github/forks/boykopovar/Grok3API?style=social) ![Issues](https://img.shields.io/github/issues/boykopovar/Grok3API?style=social) 一个用于通过移动端 gRPC 频道与 Grok 交互的异步库:发送消息、流式响应、解析 protobuf 数据块,以及检索结构化 API 数据。 ## [➡ 俄文 README](docs/Ru/RuReadMe.md) ## 功能特性 * 异步客户端 * 通过 `new_ask_stream()` 实现流式响应 * 通过 `new_ask()` 实现常规请求 * protobuf/gRPC 帧 * 类型化的 Pydantic 模型 * 流式数据块解析 * REST/gRPC/流式错误处理 * 支持元数据/工具响应/搜索结果 ## 安装说明 ``` pip install git+ssh://git@github.com/boykopovar/Grok3API.git -U ``` ## 前置条件 * Python 3.8+ * aiohttp * coincurve ## 流式示例 ``` import asyncio from grok3api.client import GrokClient from grok3api.types.request import ChatRequest from grok3api.types import TokenChunk async def main(): async with GrokClient() as client: while True: print("\nGrok: ", end="") async for chunk in client.new_ask_stream( request=ChatRequest( message=input("\nYou: "), temporary=False ), chunks_white_list=(TokenChunk,) ): print(chunk.token, end="", flush=True) if __name__ == "__main__": asyncio.run(main()) ``` ## 标准请求 ``` import asyncio from grok3api.client import GrokClient from grok3api.types.request import ChatRequest async def main(): async with GrokClient() as client: response = await client.new_ask( ChatRequest( message="Hello", temporary=False ) ) print(response.text) print(response.model_response) if __name__ == "__main__": asyncio.run(main()) ``` ## 继续已开始的对话 ``` import asyncio from grok3api.client import GrokClient from grok3api.types.request import ChatRequest, AddResponseRequest async def main(): async with GrokClient() as client: first = await client.new_ask( ChatRequest( message="Hello", temporary=False ) ) second = await client.add_response( AddResponseRequest( conversation_id=first.conversation.conversation_id, message="How are you?" ) ) print(second.text) if __name__ == "__main__": asyncio.run(main()) ``` ## Pydantic 模型 所有模型均完全匹配实际的 protobuf/gRPC API 响应。 本库不包含: * 简化结构 * 人为的抽象层 * 字段规范化 * payload 字段重命名 * 隐藏原始 API 值 ## 支持内容 ### 响应数据块 * `TokenChunk` * `ModelResponseChunk` * `FinalMetadataChunk` * `SurveyChunk` * `ConversationChunk` * `TitleChunk` ### 响应模型 * `AskResponse` * `ModelResponse` * `FinalMetadata` * `Conversation` * `Survey` * `ResponseStep` * `ToolUsageResult` * `WebSearchResult` * `XPost` * `RagResult` * `CollectionSearchResultItem` * `ConnectorSearchResultItem` * `StreamError` ### 工具响应 * 网页搜索 * X/Twitter 搜索 * RAG 结果 * 连接器搜索 * 集合搜索 * 记忆更新 * 代码执行 * 图像生成 * 视频生成 * 音频生成 ## 错误处理 ``` from grok3api.types.exceptions import ( GrokApiError, GrokRestError, GrokGrpcError, GrokRateLimitError, GrokUnderHeavyUsageError, GrokStreamError, GrokUnavailableRegionError, GrokTooManyRequestsError ) ``` ## 客户端工作原理 1. 将 protobuf 请求序列化为二进制 payload 2. 将 payload 封装进 gRPC 帧 3. 将请求发送至移动端 gRPC endpoint 4. 客户端读取流式响应 5. 将 protobuf 数据块解码为 Pydantic 模型 ## 免责声明 本项目不隶属于 xAI,也不是官方 SDK。 这是一个通过逆向工程获得的移动端 Grok API 独立实现。
标签:aiohttp, AI对话, API客户端, coincurve, Grok, gRPC, Protobuf, Pydantic, Python, Python工具, SEO, 云资产清单, 开源库, 异步编程, 搜索引擎爬虫, 数据流, 无后门, 移动端接口, 网络信息收集, 网络抓包, 逆向工具, 逆向工程