Skip to content

Interviews (async)

Async variants of :class:~superme_sdk.services._interviews.InterviewsMixin.

Source code in superme_sdk/services/aio/_interviews.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
class AsyncInterviewsMixin:
    """Async variants of :class:`~superme_sdk.services._interviews.InterviewsMixin`."""

    async def start_interview(self, role_id: str) -> dict:
        """Start a background agent interview via REST API (async).

        Returns:
            Dict with ``interview_id`` and ``status``.
        """
        resp = await self._async_rest_http.post(
            "/api/v3/agent/interview/start",
            json={"role_id": role_id},
        )
        self._check_rest_response(resp)
        return resp.json()

    async def get_interview_status(self, interview_id: str) -> InterviewStatus:
        """Poll interview status (async).

        Returns:
            Dict with ``status``, ``stages``, and other session info.
        """
        resp = await self._async_rest_http.get(
            f"/api/v3/interview/{interview_id}/status"
        )
        self._check_rest_response(resp)
        return resp.json()

    async def get_interview_transcript(self, interview_id: str) -> dict:
        """Get the full transcript for an interview (async).

        Returns:
            Dict with ``transcript`` list of stages and messages.
        """
        resp = await self._async_rest_http.get(
            f"/api/v3/interview/{interview_id}/transcript"
        )
        self._check_rest_response(resp)
        return resp.json()

    async def list_my_interviews(self) -> list[dict]:
        """List interviews for the authenticated user (async).

        Returns:
            List of interview summary dicts.
        """
        uid = self.user_id
        if not uid:
            raise ValueError("Cannot extract user_id from token")
        resp = await self._async_rest_http.get(f"/api/v3/interview/by-user/{uid}")
        self._check_rest_response(resp)
        return resp.json().get("interviews", [])

    async def send_interview_message(
        self,
        interview_id: str,
        message: str,
        *,
        stage_number: int | None = None,
        attachments: list[dict[str, Any]] | None = None,
    ) -> dict:
        """Send a candidate message during an AWAITING_INPUT stage (async).

        Returns:
            Dict with ``interview_id``, ``stage_name``, ``message``, and
            ``next_manual_stage``.
        """
        payload: dict[str, Any] = {"message": message}
        if stage_number is not None:
            payload["stage_number"] = stage_number
        if attachments is not None:
            payload["attachments"] = attachments
        resp = await self._async_rest_http.post(
            f"/api/v3/agent/interview/{interview_id}/message",
            json=payload,
        )
        self._check_rest_response(resp)
        return resp.json()

    async def stream_interview(self, interview_id: str) -> AsyncGenerator[dict, None]:
        """Stream interview events via SSE (async).

        Example:
            ```python
            async for event in client.stream_interview("interview_abc123"):
                if event.get("event") == "message":
                    print(event["content"])
                elif event.get("event") == "status":
                    print("status:", event["status"])
            ```

        Yields dicts parsed from the SSE ``data:`` lines. Each dict has an
        ``event`` key (``"message"``, ``"status"``, or ``"stage_change"``).

        Terminal statuses (``completed``, ``scoring``, ``scored``, ``failed``,
        ``withdrawn``) cause the generator to return.
        """
        terminal = {"completed", "scoring", "scored", "failed", "withdrawn"}
        async with self._async_rest_http.stream(
            "GET",
            f"/api/v3/agent/interview/{interview_id}/stream",
            headers={"Accept-Encoding": "identity"},
            timeout=None,
        ) as resp:
            if not resp.is_success:
                await resp.aread()
            self._check_rest_response(resp)
            async for line in aiter_sse_lines(resp):
                try:
                    obj = json.loads(line)
                except (json.JSONDecodeError, ValueError):
                    continue
                if isinstance(obj, dict):
                    yield obj
                    if obj.get("status") in terminal:
                        return

start_interview async 

start_interview(role_id: str) -> dict

Start a background agent interview via REST API (async).

Returns:

Type Description
dict

Dict with interview_id and status.
Source code in superme_sdk/services/aio/_interviews.py 
15
16
17
18
19
20
21
22
23
24
25
26
async def start_interview(self, role_id: str) -> dict:
    """Start a background agent interview via REST API (async).

    Returns:
        Dict with ``interview_id`` and ``status``.
    """
    resp = await self._async_rest_http.post(
        "/api/v3/agent/interview/start",
        json={"role_id": role_id},
    )
    self._check_rest_response(resp)
    return resp.json()

get_interview_status async 

get_interview_status(interview_id: str) -> InterviewStatus

Poll interview status (async).

Returns:

Type Description
InterviewStatus

Dict with status, stages, and other session info.
Source code in superme_sdk/services/aio/_interviews.py 
28
29
30
31
32
33
34
35
36
37
38
async def get_interview_status(self, interview_id: str) -> InterviewStatus:
    """Poll interview status (async).

    Returns:
        Dict with ``status``, ``stages``, and other session info.
    """
    resp = await self._async_rest_http.get(
        f"/api/v3/interview/{interview_id}/status"
    )
    self._check_rest_response(resp)
    return resp.json()

get_interview_transcript async 

get_interview_transcript(interview_id: str) -> dict

Get the full transcript for an interview (async).

Returns:

Type Description
dict

Dict with transcript list of stages and messages.
Source code in superme_sdk/services/aio/_interviews.py 
40
41
42
43
44
45
46
47
48
49
50
async def get_interview_transcript(self, interview_id: str) -> dict:
    """Get the full transcript for an interview (async).

    Returns:
        Dict with ``transcript`` list of stages and messages.
    """
    resp = await self._async_rest_http.get(
        f"/api/v3/interview/{interview_id}/transcript"
    )
    self._check_rest_response(resp)
    return resp.json()

list_my_interviews async 

list_my_interviews() -> list[dict]

List interviews for the authenticated user (async).

Returns:

Type Description
list[dict]

List of interview summary dicts.
Source code in superme_sdk/services/aio/_interviews.py 
52
53
54
55
56
57
58
59
60
61
62
63
async def list_my_interviews(self) -> list[dict]:
    """List interviews for the authenticated user (async).

    Returns:
        List of interview summary dicts.
    """
    uid = self.user_id
    if not uid:
        raise ValueError("Cannot extract user_id from token")
    resp = await self._async_rest_http.get(f"/api/v3/interview/by-user/{uid}")
    self._check_rest_response(resp)
    return resp.json().get("interviews", [])

send_interview_message async 

send_interview_message(
    interview_id: str,
    message: str,
    *,
    stage_number: int | None = None,
    attachments: list[dict[str, Any]] | None = None,
) -> dict

Send a candidate message during an AWAITING_INPUT stage (async).

Returns:

Type Description
dict

Dict with interview_id, stage_name, message, and
dict

next_manual_stage.
Source code in superme_sdk/services/aio/_interviews.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def send_interview_message(
    self,
    interview_id: str,
    message: str,
    *,
    stage_number: int | None = None,
    attachments: list[dict[str, Any]] | None = None,
) -> dict:
    """Send a candidate message during an AWAITING_INPUT stage (async).

    Returns:
        Dict with ``interview_id``, ``stage_name``, ``message``, and
        ``next_manual_stage``.
    """
    payload: dict[str, Any] = {"message": message}
    if stage_number is not None:
        payload["stage_number"] = stage_number
    if attachments is not None:
        payload["attachments"] = attachments
    resp = await self._async_rest_http.post(
        f"/api/v3/agent/interview/{interview_id}/message",
        json=payload,
    )
    self._check_rest_response(resp)
    return resp.json()

stream_interview async 

stream_interview(
    interview_id: str,
) -> AsyncGenerator[dict, None]

Stream interview events via SSE (async).

Example 
async for event in client.stream_interview("interview_abc123"):
    if event.get("event") == "message":
        print(event["content"])
    elif event.get("event") == "status":
        print("status:", event["status"])

Yields dicts parsed from the SSE data: lines. Each dict has an event key ("message", "status", or "stage_change").

Terminal statuses (completed, scoring, scored, failed, withdrawn) cause the generator to return.

Source code in superme_sdk/services/aio/_interviews.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def stream_interview(self, interview_id: str) -> AsyncGenerator[dict, None]:
    """Stream interview events via SSE (async).

    Example:
        ```python
        async for event in client.stream_interview("interview_abc123"):
            if event.get("event") == "message":
                print(event["content"])
            elif event.get("event") == "status":
                print("status:", event["status"])
        ```

    Yields dicts parsed from the SSE ``data:`` lines. Each dict has an
    ``event`` key (``"message"``, ``"status"``, or ``"stage_change"``).

    Terminal statuses (``completed``, ``scoring``, ``scored``, ``failed``,
    ``withdrawn``) cause the generator to return.
    """
    terminal = {"completed", "scoring", "scored", "failed", "withdrawn"}
    async with self._async_rest_http.stream(
        "GET",
        f"/api/v3/agent/interview/{interview_id}/stream",
        headers={"Accept-Encoding": "identity"},
        timeout=None,
    ) as resp:
        if not resp.is_success:
            await resp.aread()
        self._check_rest_response(resp)
        async for line in aiter_sse_lines(resp):
            try:
                obj = json.loads(line)
            except (json.JSONDecodeError, ValueError):
                continue
            if isinstance(obj, dict):
                yield obj
                if obj.get("status") in terminal:
                    return