서버

서버 (Server)

AG-UI 호환 서버를 구현하는 방법을 설명드릴게요.

출처: 문서

본문

소개 (Introduction)

서버 구현을 통해 에이전트나 서버에서 AG-UI 이벤트를 직접 발행할 수 있어요. 이 접근 방식은 처음부터 새 에이전트를 만들거나 에이전트 기능을 위한 전용 서비스를 원할 때 이상적입니다.

언제 서버 구현을 사용할까 (When to use a server implementation)

서버 구현은 에이전트나 서버에서 AG-UI 이벤트를 직접 발행할 수 있게 해 줍니다. 에이전트 프레임워크를 사용하지 않거나 아직 에이전트 프레임워크용 프로토콜을 만들지 않았다면, 이것이 시작하기 가장 좋은 방법이에요.

서버 구현은 또한 다음에 좋아요:

• 처음부터 새 에이전트 프레임워크 구축
• 어떤 이벤트를 어떻게 발행할지에 대한 최대 제어
• 에이전트를 독립형 API로 노출

무엇을 만들까 (What you'll build)

이 가이드에서 우리는 다음을 하는 독립형 HTTP 서버를 만들게요:

1. AG-UI 프로토콜 요청 수용
2. OpenAI의 GPT-4o 모델에 연결
3. AG-UI 이벤트로 응답을 다시 스트리밍
4. 도구 호출과 상태 관리 처리

시작해 볼게요!

사전 요구사항 (Prerequisites)

시작하기 전에 다음이 있는지 확인하세요:

• Python 3.12 이상
• 의존성 관리를 위한 Poetry
• OpenAI API 키

1. OpenAI API 키 제공

먼저 API 키를 설정해 볼게요:

# Set your OpenAI API key
export OPENAI_API_KEY=your-api-key-here

2. 빌드 유틸리티 설치

다음 도구를 설치하세요:

brew install protobuf

npm i nx

curl -fsSL https://get.pnpm.io/install.sh | sh -

1단계 – 서버 스캐폴딩 (Step 1 – Scaffold your server)

레포를 클론하는 것으로 시작하세요:

git clone [email protected]:ag-ui-protocol/ag-ui.git
cd ag-ui

server-starter 템플릿을 복사해 OpenAI 서버를 만드세요:

cp -r integrations/server-starter integrations/openai-server

메타데이터 업데이트

integrations/openai-server/package.json을 열고 필드를 새 폴더에 맞게 업데이트하세요:

{
  "name": "@ag-ui/openai-server",
  "author": "Your Name <[email protected]>",
  "version": "0.0.1",

  ... rest of package.json
}

다음으로 integrations/openai-server/src/index.ts 안의 클래스 이름을 업데이트하세요:

// Change the name to OpenAIServerAgent to add a minimal middleware for your integration.
// You can use this later on to add configuration etc.
export class OpenAIServerAgent extends HttpAgent {}

마지막으로 apps/dojo/src/menu.ts에 추가해 dojo에 통합을 소개하세요:

// ...
export const menuIntegrations: MenuIntegrationConfig[] = [
  // ...

  {
    id: "openai-server",
    name: "OpenAI Server",
    features: ["agentic_chat"],
  },
]

그리고 apps/dojo/src/agents.ts:

// ...
import { OpenAIServerAgent } from "@ag-ui/openai-server"

export const agentsIntegrations: AgentIntegrationConfig[] = [
  // ...

  {
    id: "openai-server",
    agents: async () => {
      return {
        agentic_chat: new OpenAIServerAgent(),
      }
    },
  },
]

2단계 – 패키지를 dojo 의존성에 추가 (Step 2 – Add package to dojo dependencies)

apps/dojo/package.json을 열고 dependencies 객체에 "@ag-ui/openai-server": "workspace:*"를 추가하세요:

{
  "name": "demo-viewer",
  "version": "0.1.0",
  "private": true,
  "scripts": {
    "dev": "next dev",
    "build": "next build",
    "start": "next start",
    "lint": "next lint"
  },
  "dependencies": {
    "@ag-ui/agno": "workspace:*",
    "@ag-ui/langgraph": "workspace:*",
    "@ag-ui/mastra": "workspace:*",
    "@ag-ui/middleware-starter": "workspace:*",
    "@ag-ui/server-starter": "workspace:*",
    "@ag-ui/server-starter-all-features": "workspace:*",
    "@ag-ui/vercel-ai-sdk": "workspace:*",
    "@ag-ui/openai-server": "workspace:*"
  }
}

3단계 – dojo와 서버 시작 (Step 3 – Start the dojo and server)

이제 작업을 실제로 확인해 볼게요. 먼저 Python 서버를 시작하세요:

cd integrations/openai-server/server/python
poetry install && poetry run dev

다른 터미널에서 dojo를 시작하세요:

# Install dependencies
pnpm install

# Compile the project and run the dojo
pnpm dev

http://localhost:3000으로 이동해 드롭다운에서 OpenAI를 선택하세요. 지금은 스텁 서버가 **Hello world!**로 응답하는 것을 볼 수 있어요.

그 스텁 서버에서 일어나는 일을 정리하면:

# integrations/openai-server/server/python/example_server/__init__.py
@app.post("/")
async def agentic_chat_endpoint(input_data: RunAgentInput, request: Request):
    """Agentic chat endpoint"""
    # Get the accept header from the request
    accept_header = request.headers.get("accept")

    # Create an event encoder to properly format SSE events
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():

        # Send run started event
        yield encoder.encode(
          RunStartedEvent(
            type=EventType.RUN_STARTED,
            thread_id=input_data.thread_id,
            run_id=input_data.run_id
          ),
        )

        message_id = str(uuid.uuid4())

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

멋지네요! 이미 RunStartedEvent를 보내고 있어요 — AG-UI 호환 엔드포인트를 향한 첫 걸음이죠. 이제 유용한 일을 하게 만들어 볼게요.

기본 채팅 구현 (Implementing Basic Chat)

엔드포인트를 향상시켜 OpenAI의 API를 호출하고 AG-UI 이벤트로 응답을 다시 스트리밍해 볼게요:

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from ag_ui.core import (
    RunAgentInput,
    EventType,
    RunStartedEvent,
    RunFinishedEvent,
    TextMessageStartEvent,
    TextMessageContentEvent,
    TextMessageEndEvent
)
from ag_ui.encoder import EventEncoder
import uuid
from openai import OpenAI
import os

app = FastAPI(title="AG-UI Endpoint")

# Initialize OpenAI client
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

@app.post("/")
async def agentic_chat_endpoint(input_data: RunAgentInput, request: Request):
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():
        # Send run started event
        yield encoder.encode(
            RunStartedEvent(
                type=EventType.RUN_STARTED,
                thread_id=input_data.thread_id,
                run_id=input_data.run_id
            )
        )

        # Convert AG-UI messages to OpenAI messages format
        openai_messages = []
        for msg in input_data.messages:
            if msg.role in ["user", "system", "assistant"]:
                openai_messages.append({
                    "role": msg.role,
                    "content": msg.content or ""
                })

        # Call OpenAI with streaming enabled
        stream = client.chat.completions.create(
            model="gpt-4o",
            stream=True,
            messages=openai_messages,
        )

        # Generate a message ID for the assistant's response
        message_id = str(uuid.uuid4())

        # Send text message start event
        yield encoder.encode(
            TextMessageStartEvent(
                type=EventType.TEXT_MESSAGE_START,
                message_id=message_id,
                role="assistant"
            )
        )

        # Process the streaming response and send content events
        for chunk in stream:
            if (chunk.choices and
                len(chunk.choices) > 0 and
                chunk.choices[0].delta and
                hasattr(chunk.choices[0].delta, 'content') and
                chunk.choices[0].delta.content):

                content = chunk.choices[0].delta.content
                yield encoder.encode(
                    TextMessageContentEvent(
                        type=EventType.TEXT_MESSAGE_CONTENT,
                        message_id=message_id,
                        delta=content
                    )
                )

        # Send text message end event
        yield encoder.encode(
            TextMessageEndEvent(
                type=EventType.TEXT_MESSAGE_END,
                message_id=message_id
            )
        )

        # Send run finished event
        yield encoder.encode(
            RunFinishedEvent(
                type=EventType.RUN_FINISHED,
                thread_id=input_data.thread_id,
                run_id=input_data.run_id
            )
        )

    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )

4단계 – OpenAI를 AG-UI와 연결 (Step 4 – Bridge OpenAI with AG-UI)

스텁을 OpenAI에서 컴플리션을 스트리밍하는 실제 서버로 변환해 볼게요.

OpenAI SDK 설치

먼저 OpenAI SDK가 필요해요:

cd integrations/openai-server/server/python
poetry add openai

AG-UI 요약

AG-UI 서버는 엔드포인트를 구현하고 다음을 신호하는 일련의 이벤트를 발행해요:

• 수명주기 이벤트 (RUN_STARTED, RUN_FINISHED, RUN_ERROR)
• 콘텐츠 이벤트 (TEXT_MESSAGE_*, TOOL_CALL_*, 그 외)

스트리밍 서버 구현

이제 스텁 서버를 실제 OpenAI 통합으로 변환할 거예요. 핵심 차이는 하드코딩된 "Hello world!" 메시지를 보내는 대신 OpenAI의 API에 연결하고 AG-UI 이벤트를 통해 응답을 다시 스트리밍한다는 점이에요.

구현은 스텁과 같은 이벤트 흐름을 따르지만, OpenAI 클라이언트 초기화를 추가하고 목 응답을 실제 API 호출로 교체해요. 응답에 도구 호출이 있으면 그것도 처리해, 필요할 때 서버가 함수를 완전히 사용할 수 있게 만들어요.

import os
import uuid
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from ag_ui.core import (
    RunAgentInput,
    EventType,
    RunStartedEvent,
    RunFinishedEvent,
    RunErrorEvent,
    TextMessageChunkEvent,
    ToolCallChunkEvent,
)
from ag_ui.encoder import EventEncoder
from openai import OpenAI

app = FastAPI(title="AG-UI OpenAI Server")

# Initialize OpenAI client - uses OPENAI_API_KEY from environment
client = OpenAI()

@app.post("/")
async def agentic_chat_endpoint(input_data: RunAgentInput, request: Request):
    """OpenAI agentic chat endpoint"""
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():
        try:
            yield encoder.encode(
                RunStartedEvent(
                    type=EventType.RUN_STARTED,
                    thread_id=input_data.thread_id,
                    run_id=input_data.run_id
                )
            )

            # Call OpenAI's API with streaming enabled
            stream = client.chat.completions.create(
                model="gpt-4o",
                stream=True,
                # Convert AG-UI tools format to OpenAI's expected format
                tools=[
                    {
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": tool.parameters,
                        }
                    }
                    for tool in input_data.tools
                ] if input_data.tools else None,
                # Transform AG-UI messages to OpenAI's message format
                messages=[
                    {
                        "role": message.role,
                        "content": message.content or "",
                        # Include tool calls if this is an assistant message with tools
                        **({"tool_calls": message.tool_calls} if message.role == "assistant" and hasattr(message, 'tool_calls') and message.tool_calls else {}),
                        # Include tool call ID if this is a tool result message
                        **({"tool_call_id": message.tool_call_id} if message.role == "tool" and hasattr(message, 'tool_call_id') else {}),
                    }
                    for message in input_data.messages
                ],
            )

            message_id = str(uuid.uuid4())

            # Stream each chunk from OpenAI's response
            for chunk in stream:
                # Handle text content chunks
                if chunk.choices[0].delta.content:
                    yield encoder.encode(
                        TextMessageChunkEvent(
                            type=EventType.TEXT_MESSAGE_CHUNK,
                            message_id=message_id,
                            delta=chunk.choices[0].delta.content,
                        )
                    )
                # Handle tool call chunks
                elif chunk.choices[0].delta.tool_calls:
                    tool_call = chunk.choices[0].delta.tool_calls[0]

                    yield encoder.encode(
                        ToolCallChunkEvent(
                            type=EventType.TOOL_CALL_CHUNK,
                            tool_call_id=tool_call.id,
                            tool_call_name=tool_call.function.name if tool_call.function else None,
                            parent_message_id=message_id,
                            delta=tool_call.function.arguments if tool_call.function else None,
                        )
                    )

            yield encoder.encode(
                RunFinishedEvent(
                    type=EventType.RUN_FINISHED,
                    thread_id=input_data.thread_id,
                    run_id=input_data.run_id
                )
            )

        except Exception as error:
            yield encoder.encode(
                RunErrorEvent(
                    type=EventType.RUN_ERROR,
                    message=str(error)
                )
            )

    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )

def main():
    """Run the uvicorn server."""
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run(
        "example_server:app",
        host="0.0.0.0",
        port=port,
        reload=True
    )

if __name__ == "__main__":
    main()

내부에서 어떤 일이 일어나나요?

서버가 무엇을 하는지 분석해 볼게요:

1. 설정 – OpenAI 클라이언트를 만들고 RUN_STARTED 발행
2. 요청 – 사용자 메시지를 stream=True로 chat.completions에 전송
3. 스트리밍 – 각 청크를 TEXT_MESSAGE_CHUNK 또는 TOOL_CALL_CHUNK로 전달
4. 종료 – RUN_FINISHED (또는 문제 발생 시 RUN_ERROR) 발행

엔드포인트를 다음으로 테스트하세요:

curl -X POST http://localhost:8000/ \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "threadId": "thread_123",
    "runId": "run_456",
    "state": {},
    "messages": [
      {
        "id": "msg_1",
        "role": "user",
        "content": "Hello, how are you?"
      }
    ],
    "tools": [],
    "context": [],
    "forwardedProps": {}
  }'

이 구현은 메시지를 처리하고 실시간으로 응답을 다시 스트리밍하는 완전한 기능의 AG-UI 엔드포인트를 만듭니다.

5단계 – 서버와 채팅 (Step 5 – Chat with your server)

dojo 페이지를 새로고침하고 입력을 시작하세요. GPT-4o가 단어 단위로 실시간으로 답변을 스트리밍하는 것을 볼 수 있을 거예요.

CopilotKit 같은 도구는 이미 AG-UI를 이해하고 플러그앤플레이 React 컴포넌트를 제공해요. 이들을 서버 엔드포인트에 연결하면 풀 기능 채팅 UI를 즉시 얻을 수 있어요.

통합 공유 (Share your integration)

다른 사람이 재사용할 수 있는 커스텀 서버를 만들었나요? 커뮤니티 기여를 환영해요!

1. AG-UI 저장소를 포크
2. integrations/ 아래에 패키지를 추가. 자세한 내용과 명명 규칙은 Contributing 참고
3. 사용 사례와 설계 결정을 설명하는 풀 리퀘스트 열기

질문이 있거나 피드백이 필요하거나 아이디어를 먼저 검증하고 싶다면 GitHub Discussions 게시판에서 스레드를 시작하세요: AG-UI GitHub Discussions board.

여러분의 통합이 다음 릴리스에 포함되어 전체 AG-UI 생태계가 성장하는 데 도움이 될 수 있어요.

결론 (Conclusion)

이제 OpenAI용 완전한 기능의 AG-UI 서버와 테스트할 로컬 플레이그라운드가 생겼어요. 여기서부터 여러분은:

• 서버를 향상시키기 위해 도구 호출 추가
• 서버를 프로덕션에 배포
• AG-UI를 다른 어떤 모델이나 서비스로도 가져오기

즐거운 빌드 되세요!

더 알아보기 (Learn more)

• 클라이언트 빌드 — AG-UI 클라이언트 만들기
• AG-UI 개요 — 프로토콜 기본 개념
• Middleware — 이벤트 변환·가로채기