# Calendly API技巧使用 REST API v2 掌握 Calendly 调度自动化。该技能涵盖事件类型管理、可用性配置、预订工作流程、Webhook 集成以及用于构建无缝会议协调的自动调度模式。 ## 何时使用此技能 ### 在以下情况下使用： * 自动化面试安排工作流程 * 建立会议预订集成 * 创建循环调度系统 * 以编程方式跟踪预定事件 * 将日历与 CRM 系统集成 * 建立预约提醒 * 创建自定义预订确认流程 * 会议结束后自动执行后续序列 * Calendly 与外部日历同步 * 构建调度分析仪表板 ### 不要在以下情况下使用： * 简单的日历显示（使用Google Calendar API） * 实时视频通话（使用 Zoom/Teams API） * 复杂的资源调度（使用专门的工具） * 仅内部会议协调（使用日历应用程序） * 一次性手动调度（直接使用Calendly UI） ## 先决条件 ### Calendly API 设置 # 1. 在 https://calendly.com/integrations/api_webhooks 获取 API 凭据 # 2. 创建个人访问令牌或 OAuth 应用程序 # 3. 注意：API v2 需要某些端点的组织级访问权限 # 个人访问令牌： # - 转到集成 > API 和 Webhooks # - 生成一个新的令牌 # - 复制令牌（仅显示一次） # OAuth 2.0 应用程序： # - 访问 https://developer.calendly.com/ # - 创建 OAuth 应用程序 # - 配置重定向 URI # - 注意 client_id 和 client_secret # 所需的 OAuth 范围： # - 默认 - 基本访问 # - Organization:read - 读取组织数据 # - 组织：写入 - 管理组织 # - user:read - 读取用户配置文件 # - Schedule_link:read - 读取调度链接 # - event_type:read - 读取事件类型 # - event_type:write - 管理事件类型 # - Scheduled_event:read - 读取预定事件 # - Scheduled_event:write - 管理预定事件 # - Invitedee:read - 读取受邀者数据 # - webhook:read - 读取 webhook # - webhook:write - 管理 webhook ### Python环境设置 # 创建虚拟环境 python -m venv calendly-env 源 calendly-env/bin/activate # Linux/macOS # calendly-env\Scripts\activate # Windows # 安装依赖项 pip 安装请求 python-dotenv httpx aiohttp # 创建需求.txt 猫 > 要求.txt << 'EOF' 请求>=2.31.0 python-dotenv>=1.0.0 httpx>=0.25.0 aiohttp>=3.9.0 派丹蒂克>=2.5.0 EOF # 环境变量猫 > .env << 'EOF' CALENDLY_API_KEY=您的个人访问令牌 CALENDLY_CLIENT_ID=您的 oauth-client-id CALENDLY_CLIENT_SECRET=您的 oauth-客户端秘密 CALENDLY_WEBHOOK_SECRET=您的 webhook 签名秘密 EOF ### API 客户端设置 # 客户端.py # 关于我：带有身份验证的 Calendly API 客户端 # ABOUTME：处理请求、分页和错误处理导入操作系统导入请求从输入导入可选、字典、任何、列表从 dotenv 导入 load_dotenv 加载_dotenv() 类 CalendlyClient: """Calendly API v2 客户端""" BASE_URL =“https://api.calendly.com” def __init__(self, api_key: str = None): self.api_key = api_key 或 os.environ.get("CALENDLY_API_KEY") self.session = requests.Session() self.session.headers.update({ "授权": f"承载{self.api_key}", “内容类型”：“应用程序/json”， }) self._current_user = 无 self._organization = 无 def _请求( 自我, 方法：str，端点：str，参数：字典=无， json：字典=无， ) -> 字典: """发出 API 请求""" url = f"{self.BASE_URL}{端点}" 响应 = self.session.request( 方法=方法，网址=网址，参数=参数， json=json, ）如果响应.status_code == 429： # 速率有限 retry_after = int(response.headers.get("重试后", 60)) 引发异常（f“速率有限。{retry_after}s后重试”）响应.raise_for_status() 如果响应.status_code == 204：返回{} 返回response.json() def get(self, 端点: str, params: Dict = None) -> Dict: """获取请求""" 返回 self._request("GET", 端点, params=params) def post(self, 端点: str, json: Dict = None) -> Dict: """POST 请求""" 返回 self._request("POST", 端点, json=json) def delete(self, 端点: str) -> 字典: """删除请求""" 返回 self._request("DELETE", 端点) def 分页( 自我, 端点：str，参数：字典=无，键：str =“集合”，限制：int = 无， ) -> 列表[字典]: """对结果进行分页""" 参数 = 参数或 {} 参数[“计数”] = 100 结果=[] 而真实：响应= self.get（端点，参数） items = response.get(key, []) 结果.扩展（项目）如果 limit 且 len(结果) >= limit: 返回结果[:限制] 分页 = response.get("分页", {}) next_page_token = pagination.get("next_page_token") 如果不是 next_page_token：打破 params["page_token"] = next_page_token 返回结果 @属性 def current_user(self) -> 字典： """获取当前用户信息（缓存）""" 如果不是 self._current_user：响应= self.get(“/用户/我”) self._current_user = response.get("资源") 返回 self._current_user @属性 def user_uri(self) -> str: """获取当前用户URI""" 返回 self.current_user["uri"] @属性 def Organization_uri(self) -> str: """获取当前组织 URI""" 返回 self.current_user["current_organization"] # 全局客户端实例客户端 = CalendlyClient() ## 核心能力 ### 1\.用户和组织管理 # 用户.py # 关于我：用户和组织管理 # 关于我：检索用户个人资料、组织信息、会员资格从客户端导入客户端 def get_current_user() -> 字典： """获取当前已认证的用户""" 响应 = client.get("/users/me") user = response.get("资源", {}) 返回{ "uri": user.get("uri"), “姓名”：user.get(“姓名”), “电子邮件”：user.get(“电子邮件”), "slug": user.get("slug"), "scheduling_url": user.get("scheduling_url"), "时区": user.get("时区"), “组织”：user.get(“当前组织”), } def get_user_by_uri(user_uri: str) -> 字典： """通过 URI 获取用户""" # 从 URI 中提取 UUID uuid = user_uri.split("/")[-1] 响应 = client.get(f"/users/{uuid}") 返回response.get("资源", {}) def get_organization(organization_uri: str = None) -> dict: """获取组织详细信息""" org_uri = Organization_uri 或 client.organization_uri uuid = org_uri.split("/")[-1] 响应 = client.get(f"/organizations/{uuid}") 返回response.get("资源", {}) def list_organization_memberships( 组织_uri：str =无，电子邮件：str = 无， ) -> 列表： """列出组织成员资格""" org_uri = Organization_uri 或 client.organization_uri params = {“组织”：org_uri} 如果电子邮件：参数[“电子邮件”] = 电子邮件返回 client.paginate("/organization_memberships", params=params) def get_user_availability_schedules(user_uri: str = None) -> 列表： """获取用户的空闲时间表""" 用户 = user_uri 或 client.user_uri 参数= {“用户”：用户} 返回 client.paginate("/user_availability_schedules", params=params) def get_user_busy_times( user_uri: str = 无，开始时间：str = 无，结束时间：str = 无， ) -> 列表： """获取用户在某个日期范围内的忙碌时间时间应采用 ISO 8601 格式：2026-01-17T00:00:00Z ”“” 用户 = user_uri 或 client.user_uri 参数 = { “用户”：用户， “开始时间”：开始时间， “结束时间”：结束时间， } 响应 = client.get("/user_busy_times", params=params) 返回response.get("集合", []) 如果 __name__ == "__main__": # 获取当前用户用户 = get_current_user() print(f"用户：{用户['姓名']} ({用户['电子邮件']})") print(f"调度 URL: {user['scheduling_url']}") # 列出会员资格会员资格 = list_organization_memberships() print(f"\n组织有 {len(memberships)} 名成员") ### 2\.事件类型 # 事件类型.py # 关于：事件类型管理 # 关于我：创建、列出和配置事件类型从客户端导入客户端输入 import 可选、列表 def list_event_types( user_uri: str = 无，组织_uri：str =无，活跃：布尔=真， ) -> 列表： """列出用户或组织的所有事件类型""" 参数 = {} 如果用户_uri：参数[“用户”] = user_uri elif Organization_uri： params["组织"] = Organization_uri 其他： params["用户"] = client.user_uri 如果 active 不是 None： params["active"] = str(active).lower() 返回 client.paginate("/event_types", params=params) def get_event_type(event_type_uri: str) -> 字典： """获取事件类型详细信息""" uuid = event_type_uri.split("/")[-1] 响应 = client.get(f"/event_types/{uuid}") 返回response.get("资源", {}) def get_event_type_by_slug(slug: str, user_uri: str = None) -> 可选[dict]: """按 slug 查找事件类型""" event_types = list_event_types(user_uri=user_uri) 对于 event_types 中的 et：如果 et.get("slug") == slug: 返回等返回无 def get_available_times( 事件类型_uri：str，开始时间：str，结束时间：str， ) -> 列表： """获取事件类型的可用时间段时间应采用 ISO 8601 格式：2026-01-17T00:00:00Z ”“” 参数 = { “事件类型”：事件类型_uri， “开始时间”：开始时间， “结束时间”：结束时间， } 响应 = client.get("/event_type_available_times", params=params) 返回response.get("集合", []) def format_event_type_summary(event_type: dict) -> dict: """格式化事件类型以供显示""" 返回{ “名称”：event_type.get(“名称”), "slug": event_type.get("slug"), “持续时间”：event_type.get(“持续时间”), "scheduling_url": event_type.get("scheduling_url"), "type": event_type.get("type"), # StandardEventType, AdhocEventType "kind": event_type.get("kind"), # 单人、round_robin、集体 “活动”：event_type.get(“活动”), “描述”：event_type.get(“description_plain”), } def list_event_types_summary(user_uri: str = None) -> 列表: """获取事件类型的摘要列表""" event_types = list_event_types(user_uri=user_uri) 返回 [format_event_type_summary(et) for et in event_types] 如果 __name__ == "__main__": # 列出所有事件类型 event_types = list_event_types_summary() print("可用的事件类型：") 对于 event_types 中的 et：状态=“活动” if et[“活动”] 否则“不活动” print(f" - {et['name']} ({et['duration']} 分钟) [{status}]") print(f" 网址: {et['scheduling_url']}") # 获取事件类型的可用时间如果事件类型： et_uri = event_types[0].get("uri") 从日期时间导入日期时间，时间增量开始 = datetime.now().isoformat() + "Z" 结束 = (datetime.now() + timedelta(days=7)).isoformat() + "Z" 次数 = get_available_times(et_uri, 开始, 结束) print(f"\n可用槽位: {len(times)}") ### 3\.预定活动 #预定事件.py # 关于我：预定的事件管理 # 关于我：列出、检索和取消计划的事件从客户端导入客户端输入 import 可选、列表从日期时间导入日期时间，时间增量 def list_scheduled_events( user_uri: str = 无，组织_uri：str =无，最短开始时间：str = 无，最大开始时间：str = 无，状态：str =“活动”，邀请者电子邮件：str = 无，排序：str =“开始时间：asc”， ) -> 列表： """列出预定的事件状态：有效，已取消排序：开始时间：升序，开始时间：降序 ”“” 参数 = { “状态”：状态， “排序”：排序， } 如果用户_uri：参数[“用户”] = user_uri elif Organization_uri： params["组织"] = Organization_uri 其他： params["用户"] = client.user_uri 如果最小开始时间： params["min_start_time"] = min_start_time 如果最大开始时间：参数[“最大开始时间”] = 最大开始时间如果邀请者_电子邮件： params["invitee_email"] = informe_email 返回 client.paginate("/scheduled_events", params=params) def get_scheduled_event(event_uri: str) -> 字典： """获取安排的活动详细信息""" uuid = event_uri.split("/")[-1] 响应 = client.get(f"/scheduled_events/{uuid}") 返回response.get("资源", {}) def cancel_scheduled_event(event_uri: str, Reason: str = None) -> dict: """取消预定的活动""" uuid = event_uri.split("/")[-1] 数据 = {} 如果原因：数据[“原因”] = 原因响应 = client.post(f"/scheduled_events/{uuid}/cancellation", json=data) 返回response.get("资源", {}) def get_upcoming_events( user_uri: str = 无， days_ahead: int = 7, ) -> 列表： """获取未来 N 天即将发生的事件""" 现在 = datetime.utcnow() 结束 = 现在 + timedelta(天=days_ahead) 返回列表_预定事件（用户_uri=用户_uri， min_start_time=now.isoformat() + "Z", max_start_time=end.isoformat() + "Z", 状态=“活动”，） def get_past_events( user_uri: str = 无， days_back：int = 30， ) -> 列表： """获取最近 N 天的过去事件""" 现在 = datetime.utcnow() 开始 = 现在 - timedelta(天=days_back) 返回列表_预定事件（用户_uri=用户_uri， min_start_time=start.isoformat() + "Z", max_start_time=now.isoformat() + "Z", 状态=“活动”，排序=“开始时间：描述”，） def format_event_summary(event: dict) -> dict: """格式化事件以供显示""" 返回{ "uri": event.get("uri"), “名称”：event.get(“名称”), “开始时间”：event.get(“开始时间”), “结束时间”：event.get(“结束时间”), “状态”：event.get(“状态”), "位置": event.get("位置", {}).get("类型"), "event_type": event.get("event_type"), "guests_count": len(event.get("event_guests", [])), "取消": event.get("取消"), } def get_events_by_email(email: str, user_uri: str = None) -> 列表: """查找具有特定受邀者电子邮件的所有活动""" 返回列表_预定事件（用户_uri=用户_uri， informe_email=电子邮件，）如果 __name__ == "__main__": # 获取即将发生的事件事件 = get_upcoming_events(days_ahead=14) print(f"即将发生的事件：{len(events)}") 对于事件中的事件：摘要 = format_event_summary(事件) print(f" - {summary['name']} 在 {summary['start_time']}") # 获取特定受邀者的事件 email_events = get_events_by_email("john@example.com") print(f"\njohn@example.com 的事件: {len(email_events)}") ### 4\.受邀者 # 受邀者.py # 关于我：预定活动的受邀者管理 # 关于我：检索受邀者详细信息和自定义答案从客户端导入客户端输入 import 可选、列表 def list_invites( event_uri：str，状态：str = 无，电子邮件：str = 无， ) -> 列表： """列出预定活动的受邀者状态：有效，已取消 ”“” uuid = event_uri.split("/")[-1] 参数 = {} 如果状态：参数[“状态”] = 状态如果电子邮件：参数[“电子邮件”] = 电子邮件返回 client.paginate(f"/scheduled_events/{uuid}/invitees", params=params) def get_invitee(invitee_uri: str) -> 字典： """获取受邀者详细信息""" # 解析被邀请者 URI 以获取事件和被邀请者 UUID 部分 = informe_uri.split("/") event_uuid = 零件[-3] informee_uuid = 部分[-1] 响应 = client.get(f"/scheduled_events/{event_uuid}/invitees/{invitee_uuid}") 返回response.get("资源", {}) def get_invitee_no_show(invitee_uri: str) -> 可选[dict]: """获取受邀者的缺席状态""" 部分 = informe_uri.split("/") informee_uuid = 部分[-1] 尝试：响应 = client.get(f"/invitee_no_shows/{invitee_uuid}") 返回response.get(“资源”) 除了例外：返回无 def mark_invitee_no_show(invitee_uri: str) -> 字典： """将受邀者标记为缺席""" 响应= client.post（“/invitee_no_shows”，json = {“invitee”：invitee_uri}）返回response.get("资源", {}) def unmark_invitee_no_show(no_show_uri: str) -> bool: """删除受邀者的缺席状态""" uuid = no_show_uri.split("/")[-1] client.delete(f"/invitee_no_shows/{uuid}") 返回真 def format_invitee_summary(invitee: dict) -> dict: """设置被邀请者的显示格式""" 返回{ “uri”：invitee.get（“uri”）， “姓名”：invitee.get(“姓名”), “电子邮件”：invitede.get(“电子邮件”), “状态”：invitede.get(“状态”), “时区”：invitede.get(“时区”), “created_at”：invitede.get(“created_at”), “重新安排”：invitede.get(“重新安排”), “问题和答案”：[ { “问题”：qa.get(“问题”), “答案”：qa.get(“答案”), } 对于 informee.get("questions_and_answers", []) 中的问答 ], “跟踪”：invitede.get(“跟踪”，{})， “utm_参数”：{ “来源”：invitee.get(“utm_source”), “中”：invitee.get(“utm_medium”), “活动”：invitede.get(“utm_campaign”), }, } def get_invitee_custom_answers(invitee: dict) -> dict: """从受邀者那里提取自定义问题答案""" 答案={} 对于 informee.get("questions_and_answers", []) 中的问答：问题 = qa.get("问题") 答案 = qa.get("答案") 答案[问题] = 答案返回答案 def get_all_invitees_for_events(event_uris: list) -> 列表： """获取多个活动的受邀者""" 所有被邀请者 = [] 对于 event_uris 中的 event_uri：被邀请者 = list_invitees(event_uri) 对于受邀者中的受邀者：被邀请者[“event_uri”] = event_uri all_invitees.extend（受邀者）返回所有受邀者如果 __name__ == "__main__": 从预定事件导入 get_upcoming_events # 获取即将举行的活动及其受邀者事件 = get_upcoming_events(days_ahead=7) 对于 events[:5] 中的事件： print(f"\n事件: {事件['名称']}") 被邀请者 = list_invitees(事件["uri"]) 对于受邀者的邀请：摘要 = format_invitee_summary(inv) print(f" - {summary['name']} ({summary['email']})") 如果摘要[“问题和答案”]：总结中的质量检查[“questions_and_answers”]： print(f" Q: {qa['问题']}") print(f" A: {qa['answer']}") ### 5\.网络钩子 # webhooks.py # 关于我：Webhook 订阅管理 # 关于我：创建、管理和处理 webhook 事件从客户端导入客户端导入hmac 导入哈希库输入 import 可选、列表 def list_webhook_subscriptions( 组织_uri：str =无， user_uri: str = 无，范围：str = 无， ) -> 列表： """列出 webhook 订阅范围：组织、用户 ”“” 参数 = {} 如果组织_uri： params["组织"] = Organization_uri elif user_uri：参数[“用户”] = user_uri 其他： params["组织"] = client.organization_uri 如果范围：参数[“范围”] = 范围返回 client.paginate("/webhook_subscriptions", params=params) def create_webhook_subscription( 网址：str，事件：列表，组织_uri：str =无， user_uri: str = 无，签名密钥：str = 无， ) -> 字典： """创建 webhook 订阅事件：invitee.created、invitee.canceled、routing_form_submission.created ”“” 数据 = { “网址”：网址， “事件”：事件， } 如果组织_uri：数据[“组织”] = Organization_uri 数据[“范围”] =“组织” elif user_uri：数据[“用户”] = user_uri 数据[“范围”] =“用户” 其他：数据[“组织”] = client.organization_uri 数据[“范围”] =“组织” 如果签名密钥：数据[“签名密钥”] = 签名密钥响应 = client.post("/webhook_subscriptions", json=data) 返回response.get("资源", {}) def get_webhook_subscription(subscription_uri: str) -> 字典： """获取 webhook 订阅详细信息""" uuid = subscription_uri.split("/")[-1] 响应 = client.get(f"/webhook_subscriptions/{uuid}") 返回response.get("资源", {}) def delete_webhook_subscription(subscription_uri: str) -> bool: """删除 webhook 订阅""" uuid = subscription_uri.split("/")[-1] client.delete(f"/webhook_subscriptions/{uuid}") 返回真 def verify_webhook_signature( 有效负载：字节，签名：str，签名密钥：str，公差：int = 180， ) -> 布尔值： """验证 Calendly webhook 签名 Calendly 使用 HMAC-SHA256 进行 Webhook 签名 ”“” 导入时间 # 解析签名头 # 格式：t=时间戳,v1=签名 parts = dict(p.split("=", 1) for p in signature.split(",")) 时间戳 = int(parts.get("t", 0)) Expected_sig = parts.get("v1", "") # 检查时间戳容差 if abs(time.time() - 时间戳) > 容差：返回错误 # 计算期望的签名 signed_payload = f"{时间戳}.{payload.decode()}" 计算的_sig = hmac.new( 签名密钥.encode(), signed_payload.encode(), hashlib.sha256, ).hexdigest() 返回 hmac.compare_digest(compated_sig, Expected_sig) # Webhook 事件类型 WEBHOOK_EVENTS = { "invitee.created": "当新受邀者安排活动时", "invitee.canceled": "当被邀请者取消活动时", "routing_form_submission.created": "当提交路由表单时", } WebhookHandler 类： """Calendly webhook 事件的处理程序""" def __init__(self,signing_key:str = None): self.signing_key = 签名密钥 self.handlers = {} def on(自身，事件：str)： """注册事件处理程序的装饰器""" def 装饰器（函数）： self.handlers[事件] = func return func 返回装饰器 def 句柄(self, 有效负载: dict) -> dict: """处理传入的 webhook 事件""" 事件 = Payload.get("事件") 数据 = Payload.get("有效负载", {}) 处理程序 = self.handlers.get(事件) if handler: return handler(data) 返回{“handled”：False，“event”：事件} # webhook 处理程序示例 webhook = WebhookHandler() @webhook.on("被邀请者.创建的") def handle_new_booking(data: dict) -> dict: """处理新的预订 webhook""" 被邀请者 = data.get("被邀请者", {}) 事件 = data.get("scheduled_event", {}) 返回{ “处理”：确实， “行动”：“预订_创建”， “invitee_email”：invitee.get(“电子邮件”), "event_name": event.get("名称"), “开始时间”：event.get(“开始时间”), } @webhook.on("被邀请者.取消") def handle_cancellation(data: dict) -> dict: """处理取消 webhook""" 被邀请者 = data.get("被邀请者", {}) 取消 = informee.get("取消", {}) 返回{ "handled": True, "action": "预订_取消", “invitee_email”：invitee.get(“电子邮件”), “原因”：取消.get(“原因”), "canceled_by":取消.get("canceled_by"), } 如果 __name__ == "__main__": # List existing webhooks webhooks = list_webhook_subscriptions() print(f"活动 webhooks: {len(webhooks)}") for wh in webhooks: 打印(f" - {wh['callback_url']}") print(f" 事件：{', '.join(wh['events'])}") print(f" 范围: {wh['scope']}") # Create a new webhook new_webhook = create_webhook_subscription( url =“https://example.com/webhooks/calendly”， events=["invitee.created", "invitee.canceled"], ） print(f"\n已创建 webhook: {new_webhook['uri']}") ### 6\.调度链路和路由 # 调度.py # ABOUTME：调度链路和路由形式 # 关于我：一次性链接、路由和预订定制从客户端导入客户端从输入 import 可选 def create_single_use_link(event_type_uri: str, max_event_count: int = 1) -> dict: """Create a single-use scheduling link 这些链接仅可用于有限数量的预订 ”“” 响应 = 客户端.post( “/scheduling_links”， json={ “最大事件计数”：最大事件计数， “所有者”：event_type_uri， "owner_type": "事件类型", }, ）返回response.get("资源", {}) def get_scheduling_link(link_uri: str) -> 字典： """获取调度链接详细信息""" uuid = link_uri.split("/")[-1] 响应 = client.get(f"/scheduling_links/{uuid}") 返回response.get("资源", {}) def list_routing_forms(organization_uri: str = None) -> 列表： """列出路由表""" org = Organization_uri 或 client.organization_uri 参数 = {"组织": org} 返回 client.paginate("/routing_forms", params=params) def get_routing_form(form_uri: str) -> 字典： """获取路由表详细信息""" uuid = form_uri.split("/")[-1] 响应 = client.get(f"/routing_forms/{uuid}") 返回response.get("资源", {}) def list_routing_form_submissions( 表单 uri: 字符串, 排序：str = "created_at:desc", ) -> 列表： """列出路由表单提交""" 参数 = { “路由表单”：form_uri， “排序”：排序， } 返回 client.paginate("/routing_form_submissions", params=params) def get_routing_form_submission(submission_uri: str) -> dict: """获取路由表单提交详细信息""" uuid = Submit_uri.split("/")[-1] 响应 = client.get(f"/routing_form_submissions/{uuid}") 返回response.get("资源", {}) def build_scheduling_url( 基本网址：str，名称：str = 无，电子邮件：str = 无， utm_source：str =无， utm_medium: str = 无， utm_campaign: str = None, 自定义答案：字典=无， ) -> 字符串: """构建预填充的调度 URL custom_answers: {"a1": "answer1", "a2": "answer2"} 用于自定义问题 ”“” 从 urllib.parse 导入 urlencode、urlparse、parse_qs、urlunparse 参数 = {} 如果名称：参数[“名称”] = 名称如果电子邮件：参数[“电子邮件”] = 电子邮件如果 utm_source: 参数[“utm_source”] = utm_source 如果 utm_medium: 参数[“utm_medium”] = utm_medium 如果 utm_campaign: 参数[“utm_campaign”] = utm_campaign 如果自定义答案： params.update（自定义答案）如果没有参数：返回base_url 解析 = urlparse(base_url) query = urlencode(params) 返回 urlunparse(( 解析方案，解析的.netloc，解析的路径，解析的.params，查询，解析的.fragment, ）） def generate_interview_links( 事件类型_uri：str，候选人：名单， ) -> 列表： """为候选人生成一次性面试链接候选人：[{“姓名”：“约翰”，“电子邮件”：“john@example.com”}，...] ”“” event_type = get_event_type(event_type_uri) base_url = event_type["scheduling_url"] 链接=[] for candidate in candidates: # 创建一次性链接链接 = create_single_use_link(event_type_uri, max_event_count=1) # 构建预填充的 URL 调度_url = 构建_调度_url( base_url=链接["booking_url"], 姓名=candidate.get("姓名"), email=candidate.get("电子邮件"), utm_source="采访", utm_campaign=candidate.get("活动", "招聘"), ）链接.追加({ “候选人”：候选人， “link_uri”：链接[“uri”]， “scheduling_url”：scheduling_url， }) 返回链接 from event_types import get_event_type 如果 __name__ == "__main__": from event_types import list_event_types # 获取事件类型 event_types = list_event_types() 如果事件类型： et = 事件类型[0] # 构建预填充的 URL url = build_scheduling_url( base_url=et["scheduling_url"], 名称=“简·多伊”，电子邮件=“jane@example.com”， utm_source="电子邮件", utm_campaign="q1-outreach", ） print(f"Pre-filled URL: {url}") # 创建一次性链接 single_use = create_single_use_link(et["uri"]) print(f"单次使用预订网址：{single_use['booking_url']}") ## 集成示例 ### Slack 通知集成 # slack_integration.py # 关于我：安排 Calendly 事件时通知 Slack # ABOUTME: Webhook handler with Slack notifications 导入操作系统导入请求从烧瓶导入烧瓶，请求，jsonify 从 webhooks 导入 WebhookHandler，verify_webhook_signature 应用程序=烧瓶（__名称__） webhook = WebhookHandler(signing_key=os.environ.get("CALENDLY_WEBHOOK_SECRET")) SLACK_WEBHOOK_URL = os.environ.get("SLACK_WEBHOOK_URL") def send_slack_notification（消息：字典）： """向 Slack 发送消息""" requests.post(SLACK_WEBHOOK_URL, json=message) @webhook.on("被邀请者.创建的") def handle_new_booking(data: dict) -> dict: """Notify Slack of new booking""" 被邀请者 = data.get("被邀请者", {}) 事件 = data.get("scheduled_event", {}) event_type = data.get("event_type", {}) # 提取自定义答案答案={} 对于 informee.get("questions_and_answers", []) 中的问答：答案[qa["问题"]] = qa["答案"] # 发送 Slack 通知块= [ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": ":calendar: 新会议安排", }, }, { “类型”：“部分”， “字段”：[ {"type": "mrkdwn", "text": f"*Event:*\n{event_type.get('name')}"}, {"type": "mrkdwn", "text": f"*被邀请者:*\n{invitee.get('name')}"}, {"type": "mrkdwn", "text": f"*Email:*\n{invitee.get('email')}"}, {"type": "mrkdwn", "text": f"*时间:*\n{event.get('start_time')}"}, ], }, ] 如果回答： answer_text = "\n".join(f"*{q}:* {a}" for q, a inanswers.items()) 块.追加（{ “类型”：“部分”， "text": {"type": "mrkdwn", "text": f"*响应:*\n{answer_text}"}, }) 块.追加（{ “类型”：“动作”， “元素”：[ { “类型”：“按钮”， "text": {"type": "plain_text", "text": "在 Calendly 中查看"}, "url": f"https://calendly.com/app/scheduled_events/{event['uri'].split('/')[-1]}", }, ], }) send_slack_notification({"blocks": blocks}) 返回 {"handled": True, "notified": "slack"} @webhook.on("被邀请者.取消") def handle_cancellation(data: dict) -> dict: """通知 Slack 取消""" 被邀请者 = data.get("被邀请者", {}) 事件 = data.get("scheduled_event", {}) 取消 = informee.get("取消", {}) 发送_slack_通知（{ “块”：[ { “类型”：“标题”， “文本”：{ “类型”：“纯文本”， "text": ":x: Meeting Canceled", }, }, { “类型”：“部分”， “字段”：[ {"type": "mrkdwn", "text": f"*Event:*\n{event.get('name')}"}, {"type": "mrkdwn", "text": f"*Invitee:*\n{invitee.get('name')}"}, {"type": "mrkdwn", "text": f"*Reason:*\n{cancellation.get('reason', 'Not provided')}"}, {"type": "mrkdwn", "text": f"*取消者:*\n{cancellation.get('canceled_by')}"}, ], }, ], }) 返回 {"handled": True, "notified": "slack"} @app.route("/webhooks/calendly", methods=["POST"]) def calendly_webhook(): """处理 Calendly webhook""" # 验证签名签名 = request.headers.get("Calendly-Webhook-Signature") 如果签名： Signing_key = os.environ.get("CALENDLY_WEBHOOK_SECRET") 如果不是 verify_webhook_signature(request.data,signature,signing_key): return jsonify({"error": "无效签名"}), 401 负载 = 请求.json 结果 = webhook.handle(有效负载) 返回 jsonify(结果) 如果 __name__ == "__main__": 应用程序.运行（端口=8080） ### GitHub 操作集成 # .github/workflows/calendly-sync.yml 名称：同步日历事件于：时间表： - cron: '0 8 * * *' # 每天上午 8 点工作流程_调度：职位：同步事件：运行：ubuntu-latest 步骤： - 使用：actions/checkout@v4 - 名称：设置Python 使用：actions/setup-python@v5 与： python-版本：'3.12' - 名称：安装依赖项运行：pip 安装请求 - 名称：获取即将发生的事件环境： CALENDLY_API_KEY：${{secrets.CALENDLY_API_KEY}} 运行： | 蟒蛇 << 'EOF' 导入操作系统导入请求从日期时间导入日期时间，时间增量导入 json API_KEY = os.environ["CALENDLY_API_KEY"] BASE_URL =“https://api.calendly.com” 标题= { "授权": f"承载{API_KEY}", “内容类型”：“应用程序/json”， } # 获取当前用户 user_response = requests.get(f"{BASE_URL}/users/me", headers=headers) user = user_response.json()["资源"] user_uri = 用户[“uri”] # 获取即将发生的事件现在 = datetime.utcnow() 结束 = 现在 + timedelta(天=7) 参数 = { “用户”：user_uri， "min_start_time": now.isoformat() + "Z", "max_start_time": end.isoformat() + "Z", “状态”：“活跃”， } events_response = requests.get( f“{BASE_URL}/scheduled_events”，标题=标题，参数=参数，） events = events_response.json()["集合"] print(f"找到 {len(events)} 即将发生的事件") # 保存到文件将 open("upcoming_events.json", "w") 作为 f： json.dump(事件，f，缩进=2) # 创建摘要摘要=[] 对于事件中的事件：摘要.追加（{ “名称”：事件[“名称”]， “开始时间”：事件[“开始时间”]， “状态”：事件[“状态”]， }) 将 open("events_summary.json", "w") 作为 f： json.dump(摘要，f，缩进=2) print("事件同步成功") EOF - 名称：上传事件神器使用：actions/upload-artifact@v4 与：名称：日历事件路径： | 即将到来的事件.json events_summary.json ## 最佳实践 ### 1\.速率限制 # 速率限制处理导入时间从 functools 导入包装 defrate_limit_handler(max_retries=3, base_delay=1): """用于处理 Calendly 速率限制的装饰器""" def 装饰器（函数）： @wraps(函数) def 包装器（*args，**kwargs）：对于范围内的尝试（max_retries）：尝试：返回 func(*args, **kwargs) 除了异常 e：如果 str(e) 中为“429”：延迟 = base_delay * (2 ** 尝试) print(f"速率有限，等待{延迟}s...") 时间.睡眠（延迟）其他：提高引发异常（“超出最大重试次数”）返回包装器返回装饰器 ### 2\.代币管理 # 安全令牌管理导入操作系统从 functools 导入 lru_cache @lru_cache() def get_calendly_client(): """使用安全令牌获取缓存的 Calendly 客户端""" 令牌 = os.environ.get("CALENDLY_API_KEY") 如果不是令牌：引发 ValueError("CALENDLY_API_KEY 未设置") 返回 CalendlyClient(api_key=token) # 从不记录令牌 def redact_token(text: str) -> str: 令牌 = os.environ.get("CALENDLY_API_KEY", "") if 标记和文本中的标记： return text.replace(token, "[已编辑]") 返回文本 ### 3\. Webhook 安全 # Webhook签名验证 def verify_and_process_webhook（请求）： """处理前验证 webhook 签名""" 签名 = request.headers.get("Calendly-Webhook-Signature") 如果没有签名：返回{“错误”：“缺少签名”}，401 Signing_key = os.environ.get("CALENDLY_WEBHOOK_SECRET") 如果不是 verify_webhook_signature(request.data,signature,signing_key): 返回{“错误”：“无效签名”}，401 # 处理网络钩子返回 process_webhook(request.json) ### 4\.错误处理 # 全面的错误处理类 CalendlyError（异常）： """基本 Calendly API 错误""" 通过类 RateLimitError(CalendlyError): """超出速率限制""" def __init__(self, retry_after: int): self.retry_after = 重试后 super().__init__(f"速率有限。{retry_after}s后重试") 类 NotFoundError(CalendlyError): """未找到资源""" 通过 def handler_api_error(响应): """处理 API 错误响应""" 如果响应.status_code == 429： retry_after = int(response.headers.get("重试后", 60)) 引发 RateLimitError(retry_after) elif 响应.status_code == 404：引发 NotFoundError(response.json().get("message")) 其他：响应.raise_for_status() ## 故障排除 ### 常见问题 **问题：401 未经授权** # 验证token是否有效 def verify_token(令牌: str) -> bool: 响应 = requests.get( “https://api.calendly.com/users/me”， headers={"Authorization": f"Bearer {token}"} ）返回响应.status_code == 200 **问题：没有返回任何事件** # 检查时间范围格式从日期时间导入日期时间 def format_time_for_api(dt: 日期时间) -> str: """Calendly API 的日期时间格式（带 Z 的 ISO 8601）""" return dt.strftime("%Y-%m-%dT%H:%M:%SZ") # 确保UTC时区开始 = datetime.utcnow() 格式化 = format_time_for_api(开始) **问题：Webhook 未接收事件** # 验证 webhook 订阅 def check_webhook_status(webhook_uri: str): webhook = get_webhook_subscription(webhook_uri) print(f"状态：{webhook.get('state')}") print(f"事件: {webhook.get('events')}") print(f"URL: {webhook.get('callback_url')}") # 验证 URL 是否可访问导入请求尝试：响应 = requests.post(webhook["callback_url"], json={"test": True}) print(f"可访问的 URL: {response.status_code < 500}") 除了异常 e： print(f"URL 无法访问：{e}") ### 调试命令 # 测试API认证卷曲-X GET“https://api.calendly.com/users/me”\ -H“授权：持有者$CALENDLY_API_KEY” # 列出事件类型 curl -X GET "https://api.calendly.com/event_types?user=$(curl -s -X GET https://api.calendly.com/users/me -H "授权：承载 $CALENDLY_API_KEY" | jq -r '.resource.uri')" \ -H“授权：持有者$CALENDLY_API_KEY” # 列出网络钩子 curl -X GET "https://api.calendly.com/webhook_subscriptions?organization=$(curl -s -X GET https://api.calendly.com/users/me -H "授权：承载 $CALENDLY_API_KEY" | jq -r '.resource.current_organization')" \ -H“授权：持有者$CALENDLY_API_KEY” ## 版本历史版本 |日期 |变化 ---|---|--- 1.0.0 | 2026-01-17 |具有全面 Calendly API v2 模式的初始版本 ## 资源 * [Calendly API 文档]() * [Calendly 开发者门户]() * [OAuth 2.0 指南]() * [Webhooks 指南]() * [速率限制]() * [API 变更日志]() ** ** _此技能为 Calendly 安排自动化提供生产就绪模式，从而实现无缝会议协调和预订工作流程。_

日历API

包含技能