Создание агента на OpenAI Agents SDK со стримингом ответа через веб-сокеты на Yandex Cloud Functions и API Gateway

1. Создайте папку yc-serverless-streaming-openai-agent и перейдите в нее.

2. Создайте файл function.py и вставьте в него код:

   import asyncio
   import json
   import os
   import random
   from typing import Dict, Any
   
   from agents import Agent, OpenAIProvider, Runner, RunConfig, function_tool, set_tracing_disabled
   from openai import AsyncOpenAI
   from openai.types.responses import ResponseTextDeltaEvent
   from yandex.cloud.serverless.apigateway.websocket.v1.connection_service_pb2 import SendToConnectionRequest
   from yandex.cloud.serverless.apigateway.websocket.v1.connection_service_pb2_grpc import ConnectionServiceStub
   from yandexcloud import SDK
   
   BASE_URL = os.getenv("BASE_URL")
   API_KEY = os.getenv("API_KEY")
   MODEL_NAME = os.getenv("MODEL_NAME")
   FOLDER_ID = os.environ.get('FOLDER_ID')
   
   client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)
   set_tracing_disabled(disabled=True)
   
   @function_tool
   def how_many_jokes() -> int:
       return random.randint(1, 10)
   
   # Инициализация SDK Yandex Cloud
   sdk = SDK()
   
   def get_websocket_service():
       return sdk.client(ConnectionServiceStub)
   
   def stream_to_websocket(connection_id: str, message: str):
       """Отправка сообщения в WebSocket соединение"""
       websocket_service = get_websocket_service()
       request = SendToConnectionRequest(
           connection_id=connection_id,
           type=SendToConnectionRequest.TEXT,
           data=message.encode('utf-8')
       )
       websocket_service.Send(request)
   
   async def process_stream(agent: Agent, run_config: RunConfig, connection_id: str, input_text: str):
       """Обработка стрима от агента и отправка в WebSocket"""
       result = Runner.run_streamed(
           agent,
           input=input_text,
           run_config=run_config
       )
   
       print("=== Run starting ===")
   
       async for event in result.stream_events():
           if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
               stream_to_websocket(connection_id, event.data.delta)
   
       print("=== Run complete ===")
   
   def handler(event: Dict[str, Any], context):
       """Основной обработчик Cloud Function"""
       try:
           # Получаем параметры из события
           input_text = event['body']
           request_context = event.get('requestContext')
           connection_id = request_context.get('connectionId')
   
           if not connection_id or not input_text:
               return {
                   'statusCode': 400,
                   'body': json.dumps({'error': 'Missing required parameters'})
               }
   
           # Создаем агента
           agent = Agent(
               name="Joker",
               instructions="First call the `how_many_jokes` tool, then tell that many jokes about topic from input.",
               tools=[how_many_jokes],
               model=f"gpt://{FOLDER_ID}/yandexgpt/latest",
           )
   
           run_config = RunConfig(
               model_provider=OpenAIProvider(
                   api_key=API_KEY,
                   project=FOLDER_ID,
                   base_url="https://rest-assistant.api.cloud.yandex.net/v1",
                   use_responses=True
               )
           )
   
           # Запускаем асинхронную обработку
           asyncio.run(process_stream(agent, run_config, connection_id, input_text))
   
           return {
               'statusCode': 200
           }
   
       except Exception as e:
           return {
               'statusCode': 500,
               'body': json.dumps({'error': str(e)})
           }
   

3. Создайте файл requirements.txt и вставьте в него код:

   openai-agents>=0.0.17
   yandexcloud>=0.227.0
   grpcio>=1.60.0
   protobuf>=4.25.1
   openai~=1.86.0
   

4. Создайте архив function.zip и добавьте в него файлы function.py и requirements.txt.

5. Создайте файл gateway-spec.yaml и вставьте в него код:

   openapi: 3.0.0
   info:
     title: Sample API
     version: 1.0.0
   paths:
     /:
       x-yc-apigateway-websocket-message:
         x-yc-apigateway-integration:
           payload_format_version: '0.1'
           function_id: <идентификатор_функции>
           tag: $latest
           type: cloud_functions
           service_account_id: <идентификатор_сервисного_аккаунта>
   

   После создания сервисного аккаунта и функции потребуется указать их идентификаторы.

Если у вас еще нет интерфейса командной строки Yandex Cloud (CLI), установите и инициализируйте его.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

1. Создайте сервисный аккаунт:

   yc iam service-account create --name agent-streamer-sa
   

   Результат:

   id: ajehqs5gee2e********
   folder_id: b1g681qpemb4********
   created_at: "2025-07-12T17:53:28.180991864Z"
   name: agent-streamer-sa
   

2. Назначьте роли сервисному аккаунту:

   yc resource-manager folder add-access-binding <имя_или_идентификатор_каталога> \
     --role serverless.functions.invoker \
     --subject serviceAccount:<идентификатор_сервисного_аккаунта>
   
   yc resource-manager folder add-access-binding <имя_или_идентификатор_каталога> \
     --role lockbox.payloadViewer \
     --subject serviceAccount:<идентификатор_сервисного_аккаунта>
   
   yc resource-manager folder add-access-binding <имя_или_идентификатор_каталога> \
     --role api-gateway.websocketWriter \
     --subject serviceAccount:<идентификатор_сервисного_аккаунта>
   
   yc resource-manager folder add-access-binding <имя_или_идентификатор_каталога> \
     --role ai.languageModels.user \
     --subject serviceAccount:<идентификатор_сервисного_аккаунта>
   

   Результат:

   effective_deltas:
     - action: ADD
       access_binding:
         role_id: serverless.functions.invoker
         subject:
           id: ajehqs5gee2e********
           type: serviceAccount
   
   effective_deltas:
     - action: ADD
       access_binding:
         role_id: lockbox.payloadViewer
         subject:
           id: ajehqs5gee2e********
           type: serviceAccount
   
   effective_deltas:
     - action: ADD
       access_binding:
         role_id: api-gateway.websocketWriter
         subject:
           id: ajehqs5gee2e********
           type: serviceAccount
   
   effective_deltas:
     - action: ADD
       access_binding:
         role_id: ai.languageModels.user
         subject:
           id: ajehqs5gee2e********
           type: serviceAccount
   

Чтобы создать сервисный аккаунт, воспользуйтесь методом REST API create для ресурса ServiceAccount или вызовом gRPC API ServiceAccountService/Create.

Чтобы назначить сервисному аккаунту роли serverless.functions.invoker, lockbox.payloadViewer, api-gateway.websocketWriter и ai.languageModels.user на каталог, воспользуйтесь методом REST API updateAccessBindings для ресурса Folder или вызовом gRPC API FolderService/UpdateAccessBindings.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Выполните команду:

yc iam api-key create \
  --service-account-id <идентификатор_сервисного_аккаунта> \
  --scopes yc.ai.languageModels.execute

Где:

• --service-account-id — идентификатор сервисного аккаунта agent-streamer-sa.
• --scopes — области действия ключа.

Результат:

api_key:
  id: aje3dkdmq2qn********
  service_account_id: ajehqs5gee2e********
  created_at: "2025-07-12T18:00:46.418035313Z"
  scope: yc.ai.languageModels.execute
  scopes:
    - yc.ai.languageModels.execute
secret: AQVNw20bbQtXhfpblT04zJs8Z1wUT5rD********

Сохраните значение поля secret.

Чтобы создать API-ключ, воспользуйтесь методом REST API create для ресурса ApiKey или вызовом gRPC API ApiKeyService/Create.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

Выполните команду:

yc lockbox secret create \
  --name api-key-secret \
  --payload "[{'key': 'api-key', 'text_value': '<секретный_ключ>'}]"

Где text_value — секретный ключ API-ключа, полученный на предыдущем шаге.

Результат:

id: e6q0rdjdggjp********
folder_id: b1g681qpemb4********
created_at: "2025-07-12T18:23:49.844Z"
name: api-key-secret
status: ACTIVE
current_version:
  id: e6qbp772i014********
  secret_id: e6q0rdjdggjp********
  created_at: "2025-07-12T18:23:49.844Z"
  status: ACTIVE
  payload_entry_keys:
    - api-key

Чтобы создать секрет, воспользуйтесь методом REST API create для ресурса Secret или вызовом gRPC API SecretService/Create.

По умолчанию используется каталог, указанный при создании профиля CLI. Чтобы изменить каталог по умолчанию, используйте команду yc config set folder-id <идентификатор_каталога>. Также для любой команды вы можете указать другой каталог с помощью параметров --folder-name или --folder-id.

1. Создайте функцию:

   yc serverless function create --name agent-streamer
   

   Результат:

   id: d4eem33cun2k********
   folder_id: b1g681qpemb4********
   created_at: "2025-07-12T18:15:59.854Z"
   name: agent-streamer
   http_invoke_url: https://functions.yandexcloud.net/d4eem33cun2k********
   status: ACTIVE
   

2. Создайте версию функции:

   yc serverless function version create \
     --function-name agent-streamer \
     --runtime python312 \
     --entrypoint function.handler \
     --memory 512m \
     --execution-timeout 30s \
     --source-path <путь_к_архиву> \
     --service-account-id <идентификатор_сервисного_аккаунта> \
     --environment BASE_URL=https://llm.api.cloud.yandex.net/v1 \
     --environment MODEL_NAME=gpt://<идентификатор_каталога>/yandexgpt/latest \
     --environment FOLDER_ID=<идентификатор_каталога> \
     --secret name=api-key-secret,key=api-key,environment-variable=API_KEY
   

   Где:

   • --source-path — путь к архиву function.zip.

   • --service-account-id — идентификатор сервисного аккаунта agent-streamer-sa.

   • --environment — переменные окружения:

     • BASE_URL — URL сервиса Yandex AI Studio: https://llm.api.cloud.yandex.net/v1.

     • MODEL_NAME — URI модели генерации текста Yandex AI Studio.

       Например: gpt://<идентификатор_каталога>/yandexgpt/latest, где <идентификатор_каталога> — идентификатор каталога, в котором вы создаете инфраструктуру.

     • FOLDER_ID — идентификатор каталога, в котором вы создаете инфраструктуру.

   • --secret — секрет api-key-secret.

   Результат:

   id: d4e6d89oi342********
   function_id: d4eem33cun2k********
   created_at: "2025-07-12T18:27:29.887Z"
   runtime: python312
   entrypoint: function.handler
   resources:
     memory: "536870912"
   execution_timeout: 30s
   service_account_id: ajehqs5gee2e********
   image_size: "59891712"
   status: ACTIVE
   tags:
     - $latest
   environment:
     BASE_URL: https://llm.api.cloud.yandex.net/foundationModels/v1
     FOLDER_ID: b1g681qpemb4********
     MODEL_NAME: gpt://b1g681qpemb4********/yandexgpt/latest
   secrets:
     - id: e6q0rdjdggjp********
       version_id: e6qbp772i014********
       key: api-key
       environment_variable: API_KEY
   log_options:
     folder_id: b1g681qpemb4********
   concurrency: "1"
   

Чтобы создать функцию, воспользуйтесь методом REST API create для ресурса Function или вызовом gRPC API FunctionService/Create.

Чтобы создать версию функции, воспользуйтесь методом REST API createVersion для ресурса Function или вызовом gRPC API FunctionService/CreateVersion.