이벤트는 에이전트가 작업하는 동안 어떤 일이 일어나는지 알려 줍니다. 항목은 나중에 조회할 수 있도록 저장된 메시지와 도구 호출입니다. 애플리케이션을 실시간으로 업데이트할 때는 이벤트를 사용하고, 저장된 기록을 표시할 때는 항목을 사용하세요.
애플리케이션은 입력 이벤트를 보내 메시지를 제출하거나, 턴을 취소하거나, 도구 실행 결과를 반환합니다. 에이전트는 출력과 세션의 변경 사항을 알리는 이벤트를 보냅니다. 입력을 보내는 방법은 세션 실행 및 이어가기를 참고하세요.
애플리케이션이 턴의 초기 이벤트부터 받을 수 있도록 작업을 보내기 전에 스트림을 구독하세요. API 클라이언트, 해당 대화의 세션 ID, 이벤트 핸들러를 전달하세요.
1
2
3
4
5
6
7
8
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// Pass your saved session ID to this helper.
async function streamSession(client, sessionId, handleEvent) {
const events = await client.beta.agents.sessions.events.stream(sessionId);
try {
for await (const event of events) {
await handleEvent(event);
switch (event.type) {
case "agent.session.idle":
continue;
case "error":
throw new Error(event.error.message);
case "agent.session.failed":
case "agent.session.environment.failed":
throw new Error(`Agent lifecycle failure: ${event.type}`);
case "agent.session.turn.failed":
if (event.turn.subagent_id === null) {
throw new Error(
`${event.type}: ${event.turn.error?.message ?? ""}`
);
}
break;
case "agent.session.turn.cancelled":
if (event.turn.subagent_id === null) {
throw new Error("The agent turn was cancelled");
}
break;
case "agent.session.turn.completed":
if (event.turn.subagent_id === null) return;
break;
}
}
throw new Error(
"Stream closed before a turn ended. Retrieve the saved state."
);
} finally {
events.controller.abort();
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23# Pass your saved session ID to this helper.
def stream_session(client: OpenAI, session_id: str, handle_event):
with client.beta.agents.sessions.events.stream(session_id) as events:
for event in events:
handle_event(event)
match event.type:
case "agent.session.idle":
continue
case "error":
raise RuntimeError(event.error.message)
case "agent.session.failed" | "agent.session.environment.failed":
raise RuntimeError(f"Agent lifecycle failure: {event.type}")
case "agent.session.turn.failed":
if event.turn.subagent_id is None:
detail = event.turn.error.message if event.turn.error else ""
raise RuntimeError(f"{event.type}: {detail}")
case "agent.session.turn.cancelled":
if event.turn.subagent_id is None:
raise RuntimeError("The agent turn was cancelled")
case "agent.session.turn.completed":
if event.turn.subagent_id is None:
return
raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29// Pass your saved session ID to this helper.
func streamSession(ctx context.Context, client *openai.Client, sessionID string, handleEvent func(openai.AgentSessionEventUnion)) error {
events := client.Beta.Agents.Sessions.Events.StreamStreaming(ctx, sessionID)
defer events.Close()
for events.Next() {
event := events.Current()
handleEvent(event)
switch event.Type {
case "agent.session.idle":
continue
case "error":
return fmt.Errorf("agent error: %s", event.RawJSON())
case "agent.session.failed", "agent.session.environment.failed":
return fmt.Errorf("agent lifecycle failure: %s", event.RawJSON())
case "agent.session.turn.failed", "agent.session.turn.cancelled":
if event.Turn.SubagentID == "" {
return fmt.Errorf("agent turn did not complete: %s", event.RawJSON())
}
case "agent.session.turn.completed":
if event.Turn.SubagentID == "" {
return nil
}
}
}
if err := events.Err(); err != nil {
return err
}
return fmt.Errorf("stream closed before a turn ended; retrieve the saved state")
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30// Pass your saved session ID to this helper.
public static void streamSession(
OpenAIClient client, String sessionId, Consumer<AgentSessionEvent> handleEvent) {
try (StreamResponse<AgentSessionEvent> events =
client.beta().agents().sessions().events().streamStreaming(sessionId)) {
var iterator = events.stream().iterator();
while (iterator.hasNext()) {
var event = iterator.next();
handleEvent.accept(event);
if (event.idle().isPresent()) {
continue;
}
if (event.error().isPresent()) {
throw new IllegalStateException("Agent error: " + event);
}
if (event.failed().isPresent() || event.environmentFailed().isPresent()) {
throw new IllegalStateException("Agent lifecycle failure: " + event);
}
if (event.turnFailed().filter(e -> e.turn().subagentId().isEmpty()).isPresent()
|| event.turnCancelled().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
throw new IllegalStateException("Agent turn did not complete: " + event);
}
if (event.turnCompleted().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
return;
}
}
throw new IllegalStateException(
"Stream closed before a turn ended. Retrieve the saved state.");
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26# Pass your saved session ID to this helper.
def stream_session(client, session_id, &handle_event)
events = client.beta.agents.sessions.events.stream_streaming(session_id)
begin
events.each do |event|
handle_event.call(event)
case event.type.to_s
when "agent.session.idle"
next
when "error"
raise event.error.message
when "agent.session.failed", "agent.session.environment.failed"
raise "Agent lifecycle failure: #{event.type}"
when "agent.session.turn.failed"
raise "#{event.type}: #{event.turn.error&.message}" if event.turn.subagent_id.nil?
when "agent.session.turn.cancelled"
raise "The agent turn was cancelled" if event.turn.subagent_id.nil?
when "agent.session.turn.completed"
return nil if event.turn.subagent_id.nil?
end
end
raise "Stream closed before a turn ended. Retrieve the saved state."
ensure
events.close
end
end
1
2
3
4
5curl -N \
"https://api.openai.com/v1/agents/sessions/$session_id/events?stream=true" \
-H "OpenAI-Beta: agents=v1" \
-H "Authorization: Bearer $OPENAI_API_KEY" \
-H "Accept: text/event-stream"
헬퍼는 각 이벤트를 핸들러에 전달한 뒤 주요 이벤트 유형을 확인합니다. agent.session.idle 이벤트를 받아도 계속 수신하며, 루트 턴이 완료되면 반환합니다. 루트 턴이 실패하거나 취소된 경우, 세션이나 환경에 장애가 발생한 경우, 또는 error 이벤트를 받은 경우에는 오류를 발생시킵니다. 하위 에이전트의 턴 이벤트로는 스트림이 종료되지 않습니다. 출력을 어떻게 표시할지는 핸들러가 결정하고, 헬퍼에서 발생한 오류는 호출자가 처리합니다. 턴이 끝나기 전에 스트림이 닫히면 헬퍼가 오류를 발생시킵니다. 연결이 끊긴 스트림 복구하기를 참고하세요.
구독 후 메시지 보내기
다음 예제에서는 메시지를 인자로 받아 스트림을 연 뒤에 전송합니다.
1
2
3
4
5
6
7
8
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// Pass your saved session ID and message to this helper.
async function sendAndStream(client, sessionId, text, handleEvent) {
const events = await client.beta.agents.sessions.events.stream(sessionId);
try {
await client.beta.agents.sessions.events.create(sessionId, {
events: [
{
type: "agent.session.input.message",
input: [{ role: "user", content: [{ type: "input_text", text }] }],
},
],
});
for await (const event of events) {
await handleEvent(event);
switch (event.type) {
case "agent.session.idle":
continue;
case "error":
throw new Error(event.error.message);
case "agent.session.failed":
case "agent.session.environment.failed":
throw new Error(`Agent lifecycle failure: ${event.type}`);
case "agent.session.turn.failed":
if (event.turn.subagent_id === null) {
throw new Error(
`${event.type}: ${event.turn.error?.message ?? ""}`
);
}
break;
case "agent.session.turn.cancelled":
if (event.turn.subagent_id === null) {
throw new Error("The agent turn was cancelled");
}
break;
case "agent.session.turn.completed":
if (event.turn.subagent_id === null) return;
break;
}
}
throw new Error(
"Stream closed before a turn ended. Retrieve the saved state."
);
} finally {
events.controller.abort();
}
}
1
2
3
4
5
6
7
8
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# Pass your saved session ID and message to this helper.
def send_and_stream(client: OpenAI, session_id: str, text, handle_event):
with client.beta.agents.sessions.events.stream(session_id) as events:
client.beta.agents.sessions.events.create(
session_id,
events=[
{
"type": "agent.session.input.message",
"input": [
{
"role": "user",
"content": [{"type": "input_text", "text": text}],
}
],
}
],
)
for event in events:
handle_event(event)
match event.type:
case "agent.session.idle":
continue
case "error":
raise RuntimeError(event.error.message)
case "agent.session.failed" | "agent.session.environment.failed":
raise RuntimeError(f"Agent lifecycle failure: {event.type}")
case "agent.session.turn.failed":
if event.turn.subagent_id is None:
detail = event.turn.error.message if event.turn.error else ""
raise RuntimeError(f"{event.type}: {detail}")
case "agent.session.turn.cancelled":
if event.turn.subagent_id is None:
raise RuntimeError("The agent turn was cancelled")
case "agent.session.turn.completed":
if event.turn.subagent_id is None:
return
raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")
1
2
3
4
5
6
7
8
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// Pass your saved session ID and message to this helper.
func sendAndStream(ctx context.Context, client *openai.Client, sessionID string, text string, handleEvent func(openai.AgentSessionEventUnion)) error {
events := client.Beta.Agents.Sessions.Events.StreamStreaming(ctx, sessionID)
defer events.Close()
if err := events.Err(); err != nil {
return err
}
err := client.Beta.Agents.Sessions.Events.New(ctx,
sessionID,
openai.BetaAgentSessionEventNewParams{
Events: []openai.AgentSessionInputParamUnion{
{
OfParamAgentSessionInputMessage: &openai.AgentSessionInputParamAgentSessionInputMessage{
Input: []openai.AgentSessionInputMessageParam{
{
Content: []openai.InputContentParamUnion{
{
OfParamInputText: &openai.InputContentParamInputText{
Text: text,
},
},
},
},
},
},
},
},
})
if err != nil {
return err
}
for events.Next() {
event := events.Current()
handleEvent(event)
switch event.Type {
case "agent.session.idle":
continue
case "error":
return fmt.Errorf("agent error: %s", event.RawJSON())
case "agent.session.failed", "agent.session.environment.failed":
return fmt.Errorf("agent lifecycle failure: %s", event.RawJSON())
case "agent.session.turn.failed", "agent.session.turn.cancelled":
if event.Turn.SubagentID == "" {
return fmt.Errorf("agent turn did not complete: %s", event.RawJSON())
}
case "agent.session.turn.completed":
if event.Turn.SubagentID == "" {
return nil
}
}
}
if err := events.Err(); err != nil {
return err
}
return fmt.Errorf("stream closed before a turn ended; retrieve the saved state")
}
1
2
3
4
5
6
7
8
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// Pass your saved session ID and message to this helper.
public static void sendAndStream(
OpenAIClient client, String sessionId, String text, Consumer<AgentSessionEvent> handleEvent) {
try (StreamResponse<AgentSessionEvent> events =
client.beta().agents().sessions().events().streamStreaming(sessionId)) {
client
.beta()
.agents()
.sessions()
.events()
.create(
EventCreateParams.builder()
.sessionId(sessionId)
.addEvent(
AgentSessionInputParam.AgentSessionInputMessage.builder()
.addInput(
AgentSessionInputMessageParam.builder()
.addInputTextContent(text)
.build())
.build())
.build());
var iterator = events.stream().iterator();
while (iterator.hasNext()) {
var event = iterator.next();
handleEvent.accept(event);
if (event.idle().isPresent()) {
continue;
}
if (event.error().isPresent()) {
throw new IllegalStateException("Agent error: " + event);
}
if (event.failed().isPresent() || event.environmentFailed().isPresent()) {
throw new IllegalStateException("Agent lifecycle failure: " + event);
}
if (event.turnFailed().filter(e -> e.turn().subagentId().isEmpty()).isPresent()
|| event.turnCancelled().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
throw new IllegalStateException("Agent turn did not complete: " + event);
}
if (event.turnCompleted().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
return;
}
}
throw new IllegalStateException(
"Stream closed before a turn ended. Retrieve the saved state.");
}
}
1
2
3
4
5
6
7
8
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# Pass your saved session ID and message to this helper.
def send_and_stream(client, session_id, text, &handle_event)
events = client.beta.agents.sessions.events.stream_streaming(session_id)
begin
client.beta.agents.sessions.events.create(
session_id,
events: [
{
type: "agent.session.input.message",
input: [
{
role: "user",
content: [
{
type: "input_text",
text: text
}
]
}
]
}
]
)
events.each do |event|
handle_event.call(event)
case event.type.to_s
when "agent.session.idle"
next
when "error"
raise event.error.message
when "agent.session.failed", "agent.session.environment.failed"
raise "Agent lifecycle failure: #{event.type}"
when "agent.session.turn.failed"
raise "#{event.type}: #{event.turn.error&.message}" if event.turn.subagent_id.nil?
when "agent.session.turn.cancelled"
raise "The agent turn was cancelled" if event.turn.subagent_id.nil?
when "agent.session.turn.completed"
return nil if event.turn.subagent_id.nil?
end
end
raise "Stream closed before a turn ended. Retrieve the saved state."
ensure
events.close
end
end
이벤트의 type에 따라 애플리케이션의 동작을 결정하세요.
- 텍스트 표시:
agent.session.turn.output_text.delta 이벤트로 받은 텍스트를 해당 콘텐츠 부분의 끝에 추가하세요. agent.session.turn.output_text.done 이벤트를 받으면 해당 부분을 이벤트에 담긴 전체 텍스트로 교체하세요. 델타 이벤트가 없을 수도 있습니다.
- 작업 진행 상황 추적: 세션, 턴, 항목 이벤트는 작업의 진행 상황을 알려 줍니다.
agent.session.turn.completed, agent.session.turn.failed, agent.session.turn.cancelled 중 어떤 이벤트가 발생했는지 확인해 턴의 결과를 판단하세요.
- 필요한 입력 제공:
agent.session.requires_action 이벤트를 받으면 세션을 조회하고 required_actions를 확인하세요. 코드에서 함수 실행 결과를 반환하거나, 환경을 연결하거나, 브라우저의 출처별 접근 요청 또는 로그인을 처리해야 할 수 있습니다.
세션이 유휴 상태이거나 스트림이 닫혔다는 사실만으로는 작업의 성공 여부를 판단할 수 없습니다. 턴이 완료되었더라도 모든 도구 호출이 성공했다고 단정할 수는 없습니다. 에이전트의 출력을 확인하세요.
item_id, output_index, content_index를 사용해 같은 콘텐츠 부분에 해당하는 텍스트 업데이트를 연결하세요. 예를 들어 아래에 간략히 표시한 두 이벤트는 같은 부분을 업데이트합니다.
1234567{
"type": "agent.session.turn.output_text.delta",
"item_id": "msg_789",
"output_index": 0,
"content_index": 0,
"delta": "Acme competes"
}
1234567{
"type": "agent.session.turn.output_text.done",
"item_id": "msg_789",
"output_index": 0,
"content_index": 0,
"text": "Acme competes on price and distribution."
}
각 이벤트에는 고유한 event_id가 있습니다. 두 이벤트의 item_id는 같은 저장 항목을 가리킵니다. 이 항목에는 메시지의 내용, 상태, 단계가 포함됩니다. 저장된 작업 조회하기를 참고하세요.
전체 이벤트 유형과 필드는 스트리밍 이벤트 레퍼런스에서 확인하세요. 스트림 이벤트는 웹훅과 별개입니다. 하위 에이전트의 활동과 명령어를 실행한 에이전트를 확인하는 방법은 위임된 작업 관찰하기를 참고하세요.
애플리케이션의 대화 상태에 있는 세션 ID를 사용해 저장된 작업을 조회하세요.
목록 조회 엔드포인트는 한 번에 한 페이지씩 반환합니다. 추가 결과를 조회하려면 SDK의 페이지네이션 헬퍼나 after 커서를 사용하세요. 한 턴의 항목이 여러 페이지에 걸쳐 있을 수 있습니다. 오래된 항목부터 읽으려면 order: "asc"를 사용하세요.
저장된 항목에서 놓친 출력을 복구할 수 있습니다. 다음 단계에 따라 애플리케이션 화면을 복원하세요.
- 새 스트림을 열고 수신되는 이벤트를 버퍼에 저장하세요.
- 스트림 연결을 유지한 채 세션과 저장된 항목을 조회하세요.
- 항목 ID를 키로 사용해 조회한 항목에서 로컬 상태를 복원하세요.
item_id를 기준으로 버퍼에 저장한 업데이트를 해당 항목에 적용하세요. 조회한 기록에서 이미 최종 상태에 도달한 항목의 업데이트는 폐기하세요.- 실시간 이벤트 처리를 재개하세요.
실패한 세션에 다시 연결하면 스트림은 저장된 실패 정보를 전달한 뒤 닫힙니다. 조회한 세션의 상태가 status: "failed"이거나 agent.session.failed 이벤트를 받으면 재연결 시도를 중단하고 세션 복구 지침을 따르세요.
조회한 세션의 required_actions를 사용해 입력을 기다리는 폼을 복원하세요. 과거 항목만으로는 아직 응답이 필요한 요청을 식별할 수 없습니다. 브라우저 승인 요청은 request_id를 기준으로 폼과 연결하고, 더 이상 대기 중이 아닌 요청의 폼은 제거하세요. 다시 연결한 후 작업이나 이전 승인 정보를 다시 보낼 필요는 없습니다.
output_text.done 이벤트를 받으면 임시 버퍼의 텍스트를 전체 텍스트로 교체할 수 있습니다. 저장된 항목으로 완료된 작업을 복구할 수 있지만, 중간에 놓친 이벤트까지 모두 복구할 수는 없습니다.
