Skip to content

Interviews

Source code in superme_sdk/services/_interviews.py 
  9
 10
 11
 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
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class InterviewsMixin:
    def start_interview(self, role_id: str) -> dict:
        """Start a background agent interview via REST API.

        Example:
            ```python
            session = client.start_interview("role_abc123")
            interview_id = session["interview_id"]
            ```

        Returns:
            Dict with ``interview_id`` and ``status`` (initially ``"preparing"``).
            Poll :meth:`get_interview_status` for progress.
        """
        resp = self._rest_http.post(
            "/api/v3/agent/interview/start",
            json={"role_id": role_id},
        )
        self._check_rest_response(resp)
        return resp.json()

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

        Example:
            ```python
            status = client.get_interview_status("interview_abc123")
            print(status["status"])  # "preparing", "in_progress", "completed", ...
            ```

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

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

        Example:
            ```python
            transcript = client.get_interview_transcript("interview_abc123")
            for stage in transcript["transcript"]:
                print(stage)
            ```

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

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

        Example:
            ```python
            interviews = client.list_my_interviews()
            for i in interviews:
                print(i["interview_id"], i["status"])
            ```

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

    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.

        Example:
            ```python
            reply = client.send_interview_message(
                "interview_abc123",
                "I have 5 years of experience building distributed systems.",
            )
            print(reply["message"])  # interviewer's reply
            print(reply["stage_name"])
            ```

        Args:
            interview_id: The interview session ID.
            message: The candidate's message text.
            stage_number: Optional stage number for manual-stage interviews.
            attachments: Optional list of ``{gcs_path, filename, content_type}`` dicts.

        Returns:
            Dict with ``interview_id``, ``stage_name``, ``message`` (interviewer reply),
            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 = self._rest_http.post(
            f"/api/v3/agent/interview/{interview_id}/message",
            json=payload,
        )
        self._check_rest_response(resp)
        return resp.json()

    def submit_interview(self, interview_id: str) -> dict:
        """Submit a completed interview for scoring.

        Example:
            ```python
            result = client.submit_interview("interview_abc123")
            print(result["status"])  # "submitted"
            ```

        Args:
            interview_id: The interview session ID.

        Returns:
            Dict with updated interview status.
        """
        resp = self._rest_http.post(f"/api/v3/interview/{interview_id}/submit")
        self._check_rest_response(resp)
        return resp.json()

    def send_interview_feedback(
        self,
        interview_id: str,
        stage_number: int,
        rating: int,
        comments: str,
    ) -> dict:
        """Leave feedback on an interview stage.

        Example:
            ```python
            result = client.send_interview_feedback(
                "interview_abc123",
                stage_number=1,
                rating=4,
                comments="Good questions, well-structured.",
            )
            ```

        Args:
            interview_id: The interview session ID.
            stage_number: The stage number to leave feedback on.
            rating: Rating from 1 to 5.
            comments: Feedback comments text.

        Returns:
            Dict with feedback confirmation.
        """
        resp = self._rest_http.post(
            f"/api/v3/agent/interview/{interview_id}/feedback",
            json={
                "stage_number": stage_number,
                "rating": rating,
                "comments": comments,
            },
        )
        self._check_rest_response(resp)
        return resp.json()


    def stream_interview(self, interview_id: str):
        """Stream interview events via SSE from ``GET /api/v3/agent/interview/{id}/stream``.

        Example:
            ```python
            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"}
        with self._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:
                resp.read()
            self._check_rest_response(resp)
            buf = ""
            for raw in resp.iter_text():
                buf += raw
                while "\n" in buf:
                    line, buf = buf.split("\n", 1)
                    line = line.strip()
                    if not line or line.startswith(":"):
                        continue
                    if line.startswith("data: "):
                        line = line[6:]
                    elif line.startswith("data:"):
                        line = line[5:]
                    else:
                        continue
                    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 

start_interview(role_id: str) -> dict

Start a background agent interview via REST API.

Example 
session = client.start_interview("role_abc123")
interview_id = session["interview_id"]

Returns:

Name Type Description
dict

Dict with interview_id and status (initially "preparing").
Poll dict

meth:get_interview_status for progress.
Source code in superme_sdk/services/_interviews.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def start_interview(self, role_id: str) -> dict:
    """Start a background agent interview via REST API.

    Example:
        ```python
        session = client.start_interview("role_abc123")
        interview_id = session["interview_id"]
        ```

    Returns:
        Dict with ``interview_id`` and ``status`` (initially ``"preparing"``).
        Poll :meth:`get_interview_status` for progress.
    """
    resp = self._rest_http.post(
        "/api/v3/agent/interview/start",
        json={"role_id": role_id},
    )
    self._check_rest_response(resp)
    return resp.json()

get_interview_status 

get_interview_status(interview_id: str) -> dict

Poll interview status.

Example 
status = client.get_interview_status("interview_abc123")
print(status["status"])  # "preparing", "in_progress", "completed", ...

Returns:

Type Description
dict

Dict with status, stages, and other session info.
Source code in superme_sdk/services/_interviews.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def get_interview_status(self, interview_id: str) -> dict:
    """Poll interview status.

    Example:
        ```python
        status = client.get_interview_status("interview_abc123")
        print(status["status"])  # "preparing", "in_progress", "completed", ...
        ```

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

get_interview_transcript 

get_interview_transcript(interview_id: str) -> dict

Get the full transcript for an interview.

Example 
transcript = client.get_interview_transcript("interview_abc123")
for stage in transcript["transcript"]:
    print(stage)

Returns:

Type Description
dict

Dict with transcript list of stages and messages.
Source code in superme_sdk/services/_interviews.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def get_interview_transcript(self, interview_id: str) -> dict:
    """Get the full transcript for an interview.

    Example:
        ```python
        transcript = client.get_interview_transcript("interview_abc123")
        for stage in transcript["transcript"]:
            print(stage)
        ```

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

list_my_interviews 

list_my_interviews() -> list[dict]

List interviews for the authenticated user.

Example 
interviews = client.list_my_interviews()
for i in interviews:
    print(i["interview_id"], i["status"])

Returns:

Type Description
list[dict]

List of interview summary dicts.
Source code in superme_sdk/services/_interviews.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def list_my_interviews(self) -> list[dict]:
    """List interviews for the authenticated user.

    Example:
        ```python
        interviews = client.list_my_interviews()
        for i in interviews:
            print(i["interview_id"], i["status"])
        ```

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

send_interview_message 

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.

Example 
reply = client.send_interview_message(
    "interview_abc123",
    "I have 5 years of experience building distributed systems.",
)
print(reply["message"])  # interviewer's reply
print(reply["stage_name"])

Parameters:

Name Type Description Default
interview_id str

The interview session ID.

 required
message str

The candidate's message text.

 required
stage_number int | None

Optional stage number for manual-stage interviews.

 None
attachments list[dict[str, Any]] | None

Optional list of {gcs_path, filename, content_type} dicts.

 None

Returns:

Type Description
dict

Dict with interview_id, stage_name, message (interviewer reply),
dict

and next_manual_stage.
Source code in superme_sdk/services/_interviews.py 
 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
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.

    Example:
        ```python
        reply = client.send_interview_message(
            "interview_abc123",
            "I have 5 years of experience building distributed systems.",
        )
        print(reply["message"])  # interviewer's reply
        print(reply["stage_name"])
        ```

    Args:
        interview_id: The interview session ID.
        message: The candidate's message text.
        stage_number: Optional stage number for manual-stage interviews.
        attachments: Optional list of ``{gcs_path, filename, content_type}`` dicts.

    Returns:
        Dict with ``interview_id``, ``stage_name``, ``message`` (interviewer reply),
        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 = self._rest_http.post(
        f"/api/v3/agent/interview/{interview_id}/message",
        json=payload,
    )
    self._check_rest_response(resp)
    return resp.json()

submit_interview 

submit_interview(interview_id: str) -> dict

Submit a completed interview for scoring.

Example 
result = client.submit_interview("interview_abc123")
print(result["status"])  # "submitted"

Parameters:

Name Type Description Default
interview_id str

The interview session ID.

 required

Returns:

Type Description
dict

Dict with updated interview status.
Source code in superme_sdk/services/_interviews.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def submit_interview(self, interview_id: str) -> dict:
    """Submit a completed interview for scoring.

    Example:
        ```python
        result = client.submit_interview("interview_abc123")
        print(result["status"])  # "submitted"
        ```

    Args:
        interview_id: The interview session ID.

    Returns:
        Dict with updated interview status.
    """
    resp = self._rest_http.post(f"/api/v3/interview/{interview_id}/submit")
    self._check_rest_response(resp)
    return resp.json()

send_interview_feedback 

send_interview_feedback(
    interview_id: str,
    stage_number: int,
    rating: int,
    comments: str,
) -> dict

Leave feedback on an interview stage.

Example 
result = client.send_interview_feedback(
    "interview_abc123",
    stage_number=1,
    rating=4,
    comments="Good questions, well-structured.",
)

Parameters:

Name Type Description Default
interview_id str

The interview session ID.

 required
stage_number int

The stage number to leave feedback on.

 required
rating int

Rating from 1 to 5.

 required
comments str

Feedback comments text.

 required

Returns:

Type Description
dict

Dict with feedback confirmation.
Source code in superme_sdk/services/_interviews.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def send_interview_feedback(
    self,
    interview_id: str,
    stage_number: int,
    rating: int,
    comments: str,
) -> dict:
    """Leave feedback on an interview stage.

    Example:
        ```python
        result = client.send_interview_feedback(
            "interview_abc123",
            stage_number=1,
            rating=4,
            comments="Good questions, well-structured.",
        )
        ```

    Args:
        interview_id: The interview session ID.
        stage_number: The stage number to leave feedback on.
        rating: Rating from 1 to 5.
        comments: Feedback comments text.

    Returns:
        Dict with feedback confirmation.
    """
    resp = self._rest_http.post(
        f"/api/v3/agent/interview/{interview_id}/feedback",
        json={
            "stage_number": stage_number,
            "rating": rating,
            "comments": comments,
        },
    )
    self._check_rest_response(resp)
    return resp.json()

stream_interview 

stream_interview(interview_id: str)

Stream interview events via SSE from GET /api/v3/agent/interview/{id}/stream.

Example 
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/_interviews.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def stream_interview(self, interview_id: str):
    """Stream interview events via SSE from ``GET /api/v3/agent/interview/{id}/stream``.

    Example:
        ```python
        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"}
    with self._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:
            resp.read()
        self._check_rest_response(resp)
        buf = ""
        for raw in resp.iter_text():
            buf += raw
            while "\n" in buf:
                line, buf = buf.split("\n", 1)
                line = line.strip()
                if not line or line.startswith(":"):
                    continue
                if line.startswith("data: "):
                    line = line[6:]
                elif line.startswith("data:"):
                    line = line[5:]
                else:
                    continue
                try:
                    obj = json.loads(line)
                except (json.JSONDecodeError, ValueError):
                    continue
                if isinstance(obj, dict):
                    yield obj
                    if obj.get("status") in terminal:
                        return