Parse a taskId into (tool_name, job_id, cache_key, created_at).
(task_id: str)
| 17 | |
| 18 | |
| 19 | def parse_task_id(task_id: str) -> tuple[str, str, str, datetime]: |
| 20 | """Parse a taskId into (tool_name, job_id, cache_key, created_at).""" |
| 21 | try: |
| 22 | tool_name, job_id, rest = task_id.split(":", 2) |
| 23 | cache_key, created_epoch = rest.rsplit(":", 1) |
| 24 | created_at = datetime.fromtimestamp(int(created_epoch), tz=timezone.utc) |
| 25 | except (ValueError, TypeError) as exc: |
| 26 | raise MCPError(f"Malformed taskId: {task_id!r}") from exc |
| 27 | return tool_name, job_id, cache_key, created_at |
| 28 | |
| 29 | |
| 30 | def _get_callback_manager(tool_name: str): |
no test coverage detected
searching dependent graphs…