Пошаговый процесс создания собственного инструмента для Opal AI в Goo

У меня была возможность принять участие в хакатоне Opal AI, где мы создали специальный инструмент с использованием Opal Python SDK от Optimizely.

В этой статье рассказывается, как создать инструмент на Python и безопасно развернуть его в Google Cloud Run. Цель этого инструмента — дать возможность агенту Opal мгновенно создать полностью подробную историю пользователя Azure DevOps (ADO) на основе простого запроса.

Ядро: SDK Optimizely Opal Tools

Optimizely предоставляет различные SDK для подключения к своим сервисам. Чтобы агент Opal мог использовать вашу услугу, ему сначала необходим план, определяющий, что делает ваш инструмент, какие входные данные ему нужны и как его выполнять. Этот чертеж называется Манифест инструментапредставленный через стандартизированный /discovery конечная точка.

SDK Python Opal Tools абстрагирует сложность управления этим контрактом API. Просто украсив стандартную функцию Python @toolSDK автоматически обрабатывает:

1. Создание необходимого OpenAPI-совместимого Манифест открытия в /discovery.

2. Маршрутизация входящих POST запросы к правильной функции.

3. Проверка и анализ входных данных JSON на основе вашей модели Pydantic.

Сервисный код инструмента (main.py)

Файл main.py является основной точкой входа вашего кода, и служба использует FastAPI для маршрутизации и Opal SDK для определения инструментов. Вы можете определить один экземпляр приложения, на котором размещено несколько конечных точек инструмента. В этом случае я создал одну конечную точку инструмента.

import os
import base64
import httpx
import pdb
from fastapi import FastAPI
from pydantic import BaseModel, Field
from opal_tools_sdk import ToolsService, tool
from dotenv import load_dotenv

# --- Configuration ---
# In production, load these from os.environ for security
ADO_ORG = "rightpoint"
ADO_PROJECT = "Optimizely-Opal-Challenge-2025"
load_dotenv()
ADO_PAT = os.environ.get("ADO_PAT")

tag = "opal-2025"

# Encode PAT for Basic Auth
auth_str = f":{ADO_PAT}"
b64_auth = base64.b64encode(auth_str.encode()).decode()
HEADERS = {
    "Authorization": f"Basic {b64_auth}",
    "Content-Type": "application/json-patch+json"
}

# --- App Setup ---
app = FastAPI()
# This initializes the /discovery endpoint automatically
service = ToolsService(app)

# --- Parameters Model ---
class UserStoryParams(BaseModel):
    title: str = Field(..., description="The title of the user story")
    description: str = Field(..., description="Detailed description of the user story")
    acceptance_criteria: str = Field(None, description="Acceptance criteria for the story")

# --- The Tool Definition ---
@tool(
    name="create_ado_user_story",
    description="Creates a new User Story in Azure DevOps with a title and description.",
)

async def create_ado_user_story(params: UserStoryParams):
    """
    Creates a User Story in Azure DevOps.
    """
    url = f"https://dev.azure.com/{ADO_ORG}/{ADO_PROJECT}/_apis/wit/workitems/$User%20Story?api-version=7.1"

    # Azure DevOps requires a JSON Patch document
    payload = [
        {
            "op": "add",
            "path": "/fields/System.Title",
            "value": params.title
        },
        {
            "op": "add",
            "path": "/fields/System.Description",
            "value": params.description
        },
        {
            "op": "add",
            "path": "/fields/System.Tags",
            "value": tag
        }
    ]

    if params.acceptance_criteria:
        payload.append({
            "op": "add",
            "path": "/fields/Microsoft.VSTS.Common.AcceptanceCriteria",
            "value": params.acceptance_criteria
        })

    response = None
    try:
        # Use httpx.AsyncClient for non-blocking I/O inside an async function
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(url, headers=HEADERS, json=payload)
            response.raise_for_status()
            data = response.json()

        # Return a dictionary directly. This dictionary is the FINAL return value
        # and should not be awaited by the SDK's wrapper.
        return {
            "status": "success",
            "id": data.get("id"),
            "link": data.get("_links", {}).get("html", {}).get("href"),
            "message": f"User Story #{data.get('id')} created successfully."
        }
    except httpx.RequestError as e: 
        # Handle all httpx communication errors (DNS, connection, etc.)
        response_text = "No response body available." if response is None else response.text
        print(f"HTTPX Request Error: {str(e)}nDetails: {response_text}")
        return {
            "status": "error",
            "message": f"Connection/Request Error: {str(e)}",
            "details": response_text
        }
    except Exception as e:
        # Handle all other exceptions (including raise_for_status errors)
        response_text = "N/A"
        if response is not None and response.text:
            response_text = response.text
            
        print(f"Unexpected Error: {str(e)}nResponse Body: {response_text}")
        return {
            "status": "error",
            "message": f"An unexpected error occurred: {str(e)}",
            "details": response_text
        }

# Run locally for testing
if __name__ == "__main__":
    import uvicorn
    # CRITICAL: Ensure you are running this command, which uses the uvicorn async server.
    uvicorn.run(app, host="0.0.0.0", port=8000)

При написании инструментов для высокопроизводительных облачных сред, таких как Cloud Run, важно использовать асинхронный (асинхронный) код. Это предотвращает блокировку всего процесса Python одним сетевым запросом (например, ожиданием медленного ответа ADO API), позволяя серверу обрабатывать десятки запросов одновременно.

Мы достигаем этого, определив функцию как async def и используя асинхронный HTTP-клиент, httpx.

Инструмент создает полезную нагрузку ADO JSON Patch, использует внедренный токен личного доступа (PAT) для аутентификации и выполняет асинхронный сетевой вызов.

Развертывание: использование Google Cloud Run

Чтобы сделать ваш инструмент общедоступным для Optimizely Opal, мы развертываем его как бессерверный контейнер в Google Cloud Run.

Файлы развертывания

Мы используем requirements.txt управлять зависимостями и Dockerfile для развертывания.

а) Зависимости (requirements.txt): Это сообщает приложению, какие зависимости необходимо установить для успешной работы приложения.

fastapi==0.110.0
uvicorn==0.27.1
requests==2.31.0
optimizely-opal.opal-tools-sdk
pydantic
python-dotenv

б) План контейнера (Dockerfile):

# Use the official lightweight Python image.
# https://hub.docker.com/_/python
FROM python:3.12-slim

# Allow statements and log messages to immediately appear in the Knative logs
ENV PYTHONUNBUFFERED True

# Copy local code to the container image.
ENV APP_HOME /app
WORKDIR $APP_HOME
COPY . ./

# Install production dependencies.
RUN pip install --no-cache-dir -r requirements.txt

# Run the web service on container startup. Here we use the gunicorn
# webserver, with one worker process and 8 threads.
# For environments with multiple CPU cores, increase the number of workers
# to be equal to the cores available.
# Timeout is set to 0 to disable the timeouts of the workers to allow Cloud Run to handle instance scaling.
CMD exec uvicorn main:app --host 0.0.0.0 --port $PORT

Этот контейнер обеспечивает безопасность, запуская приложение от имени пользователя, не являющегося пользователем root, и явно определяет команду запуска.

Этапы развертывания (gcloud CLI)

1. Защитите PAT: Загрузите свой PAT Azure DevOps в Google Cloud Secret Manager (рекомендуется, как показано в предыдущем контексте).

2. Сборка и развертывание: Используйте gcloud CLI для сборки контейнера из исходного кода и его развертывания, сопоставляя секрет с ADO_PAT переменная среды.