Azure Event Hubs Python 技能
为 Python 应用提供高吞吐事件流的发送、接收与检查点模式。
使用 DefaultAzureCredential、避免明文连接字符串，并建议生产环境限制凭据链，降低了凭据暴露风险；但示例直接发送、接收和检查点写入 Azure 外部服务，缺少用户确认、数据流/敏感数据说明、权限范围、回滚和操作影响提示，因此扣分。
文档覆盖生产者、消费者、分区、检查点和错误处理，且提供 CLI；但静态证据显示异步示例缺少 EventData 导入，异步凭据未按自身规则使用上下文管理器，脚本中的 StopIteration 控制流和部分 API 用法未由技能内测试验证，因此仅给有限分数。
名称、受众、触发词和主要场景较清楚，覆盖同步/异步、高吞吐、分区和检查点；但非适用边界、输入约束、输出契约和语义触发排除条件不充分，也没有中文使用说明；核心依赖 Azure 服务，网络可达性未说明，因此扣分。
Front matter 包含名称、版本、作者、包名和 MIT 许可证，主文档配有安装、环境变量、示例及分层参考文件；但依赖未固定版本，缺少变更日志、维护责任/更新路径、FAQ、兼容性矩阵和系统化故障排查，且部分示例假设未明确，因此扣分。
内容能直接指导常见 Event Hubs 生产、消费、分区和检查点任务，CLI 也提供信息查询、收发事件能力；但示例存在明显完整性问题且没有技能专属可执行验证，生成代码仍需较多人工核对，因此按静态校准限制仅给中低分。
仓库提供通用 CI、评估工作流和测试基础设施，并有可审阅的源码、文档与许可证；但给定材料没有覆盖此技能关键路径的专属测试、真实执行结果或第三方交叉验证，结论主要依赖静态阅读，因此分数有限。
- 发送命令会向 Azure Event Hubs 写入外部数据；生产使用前应明确确认目标命名空间、事件内容、权限和成本影响。
- 异步示例应补充 EventData 导入，并统一异步凭据的生命周期管理。
- 不要把文档示例直接视为已验证 API；应在锁定依赖版本的隔离环境中运行针对生产、消费、检查点和分区的测试。
- Azure 服务访问依赖凭据、RBAC、网络和区域可达性；材料未说明中国大陆网络可用性或替代方案。
它能做什么 & 适用场景
该技能面向使用 Azure Event Hubs SDK for Python 构建事件流应用的开发者。它涵盖生产者、消费者、批量发送、分区、消费组、事件属性和 Blob 检查点存储。技能优先推荐 DefaultAzureCredential，并要求客户端使用上下文管理器进行确定性资源释放。它同时提供同步和异步客户端示例，适合高吞吐事件摄取与可靠消费场景。
指导安装 azure-eventhub、azure-identity 及可选的 Blob 检查点包；配置事件中心命名空间、事件中心名称和存储账户环境变量；创建 EventHubProducerClient 与 EventHubConsumerClient；创建并发送事件批次，处理批次大小限制；按分区 ID 或分区键发送；从事件流接收事件并更新检查点；使用 BlobCheckpointStore 跟踪消费进度；读取事件属性、序列号、偏移量和入队时间；查询事件中心及分区信息；编写同步或异步客户端代码。
- 负责 Azure 数据管道的 Python 开发者，需要批量摄取大量事件到 Event Hubs。
- 构建多分区消费者的团队，需要按消费组读取事件并记录处理进度。
- 开发实时流处理服务的工程师，需要使用异步客户端提升吞吐量。
- 需要避免连接字符串和密钥的 Azure 应用团队，希望通过 DefaultAzureCredential 使用 Entra 身份认证。
- 需要可靠恢复消费位置的生产系统维护者，需要配置 Blob 检查点存储。
优缺点一览
- 覆盖生产者、消费者、批处理、分区和检查点等核心 Event Hubs 操作。
- 同时提供同步与异步客户端模式，并强调客户端和异步凭据的生命周期管理。
- 推荐 DefaultAzureCredential，支持本地开发与 Azure 环境之间的身份认证迁移。
- 包含批次大小限制、消费组和分区键等实际开发注意事项。
- 仓库采用 MIT 许可证，并通过集合安装方式提供。
- 范围仅限 Azure Event Hubs 的 Python SDK，不覆盖其他语言或 Azure 消息服务。
- 可靠消费依赖 Blob 检查点时，需要额外安装包并配置 Blob 存储环境变量。
- 提供的材料没有列出该技能专属的测试场景或测试结果。
- 参考文件的具体内容未包含在提供材料中，因此其细节无法在此核实。
如何安装
在 Agent Skills 兼容客户端中安装该仓库集合：运行 npx skills add microsoft/skills，然后在向导中选择所需技能；技能会安装到所选 Agent 的目录，例如 GitHub Copilot 使用 .github/skills/。在 Python 项目中安装依赖：pip install azure-eventhub azure-identity。需要 Blob 检查点时再运行：pip install azure-eventhub-checkpointstoreblob-aio。
如何使用
使用包含触发词的请求，例如：“帮我用 EventHubProducerClient 为 Python 服务实现批量事件发送，并使用 DefaultAzureCredential。”也可以请求消费者、分区键发送、Blob 检查点或异步 Event Hubs 客户端示例。运行代码前设置 EVENT_HUB_FULLY_QUALIFIED_NAMESPACE 和 EVENT_HUB_NAME；使用检查点时还需设置 STORAGE_ACCOUNT_URL 和 CHECKPOINT_CONTAINER。生产环境使用 DefaultAzureCredential 时设置 AZURE_TOKEN_CREDENTIALS=prod 或指定凭据。