title: "Агенты + API: Построение инструментов из OpenAPI спецификаций, динамическое создание инструментов" slug: agents-apis-tool-building-openapi-2026-ru date: 2026-02-22 lang: ru

Агенты + API: Построение инструментов из OpenAPI спецификаций, динамическое создание инструментов

Ключевые факты

• langchain v1.2.10 (последняя на март 2026) — предоставляет langchain.tools.OpenAPISpec загрузчик
• langgraph v1.0.10 — оркестрация конечного автомата для многотулов рабочих процессов
• httpx v0.28.1 — асинхронный HTTP-клиент с пулингом соединений, управлением timeout
• pydantic v2.12.5 — v2 использует model_validate(), v1 использовала parse_obj() для валидации входа инструмента
• pydantic-core построена на Rust в Pydantic v2, быстрее валидирует
• OpenAI function calling — макс 128 инструментов в одном запросе, макс 256 параметров на инструмент (GPT-4 Turbo)
• Anthropic Claude — поддерживает до 2048 инструментов в одном сообщении (Claude 3.5 Sonnet)
• OpenAPI 3.1 использует JSON Schema 2020-12, OpenAPI 3.0 использует JSON Schema draft-04 (несовместима валидация enum)
• LangChain OpenAPIToolkit — класс langchain_community.agent_toolkits.openapi.toolkit.OpenAPIToolkit
• LangChain RequestsToolkit — оборачивает HTTP операции: GET, POST, PATCH, DELETE, PUT
• Паттерны аутентификации для OpenAPI инструментов: Bearer token (OAuth2), API key в заголовке, mTLS клиентские сертификаты, Basic Auth
• Ограничения имени инструмента: Anthropic требует ^[a-zA-Z0-9_-]{1,64}$, имена OpenAI функций должны соответствовать ^[a-zA-Z0-9_-]{1,64}$
• Rate limiting: алгоритм token bucket стандартный — токены пополняются с постоянной скоростью, ёмкость burst = размер корзины
• OpenAPI operationId поле рекомендуется для стабильных имён инструментов; без него генерируется из {method}_{path_slug}
• OpenAPI $ref resolution — LangChain v1.2+ использует библиотеку jsonref для разрешения ссылок перед преобразованием схемы
• Паттерны пагинации: cursor-based (next_cursor), offset-based (offset/limit), page-token (page_token)
• HTTP 429 Retry-After заголовок — сервер возвращает время ожидания в секундах или HTTP-date формате (RFC 7231)
• Лимиты размера результата инструмента: Anthropic макс 10,000 символов на результат инструмента, OpenAI макс 100,000 токенов всего контекста
• OpenAPI security schemes: apiKey (в заголовке/query/cookie), http (bearer/basic), oauth2, openIdConnect
• Паттерн динамической регистрации инструментов — инструменты загружаются во время выполнения из URL спецификации, позволяет workflow discover-and-use

Построение инструментов из OpenAPI спецификаций

Автоматический pipeline: OpenAPI spec → schema parser → Pydantic models → LangChain tool definitions

Шаг 1: Загрузка и парсинг OpenAPI спецификации

import httpx
import yaml
import json

def load_openapi_spec(source: str) -> dict:
    """Load OpenAPI spec from URL or file path. Supports JSON and YAML."""
    if source.startswith("http"):
        response = httpx.get(source)
        response.raise_for_status()
        content = response.text
    else:
        with open(source) as f:
            content = f.read()

    try:
        return json.loads(content)
    except json.JSONDecodeError:
        return yaml.safe_load(content)

Шаг 2: Преобразование OpenAPI схемы в JSON Schema для валидации входа инструмента

def openapi_to_json_schema(openapi_schema: dict) -> dict:
    """Convert OpenAPI 3.x parameter schema to JSON Schema for Anthropic tool input."""
    schema = {}

    if "type" in openapi_schema:
        schema["type"] = openapi_schema["type"]

    if "description" in openapi_schema:
        schema["description"] = openapi_schema["description"]

    if "enum" in openapi_schema:
        schema["enum"] = openapi_schema["enum"]

    if openapi_schema.get("type") == "array" and "items" in openapi_schema:
        schema["items"] = openapi_to_json_schema(openapi_schema["items"])

    if openapi_schema.get("type") == "object" and "properties" in openapi_schema:
        schema["properties"] = {
            k: openapi_to_json_schema(v)
            for k, v in openapi_schema["properties"].items()
        }

    # Transfer constraints
    for constraint in ["minimum", "maximum", "minLength", "maxLength", "pattern"]:
        if constraint in openapi_schema:
            schema[constraint] = openapi_schema[constraint]

    return schema

Шаг 3: Генерация определений инструментов из endpoints

Извлеките метаданные endpoint (метод, путь, параметры, тело запроса) и построите совместимую с Anthropic схему инструмента.

def generate_tool_from_endpoint(method: str, path: str, operation: dict) -> dict:
    """Build a single tool definition from an OpenAPI operation."""
    name = operation.get("operationId", f"{method}_{path.replace('/', '_').strip('_')}")
    name = name.replace("-", "_")[:64]  # Anthropic name constraints

    description = operation.get("summary", f"{method.upper()} {path}")

    properties = {}
    required = []

    # Extract parameters (path, query, header)
    for param in operation.get("parameters", []):
        param_name = param["name"]
        param_schema = param.get("schema", {"type": "string"})

        properties[param_name] = openapi_to_json_schema(param_schema)

        if param.get("required", param["in"] == "path"):
            required.append(param_name)

    # Extract request body schema
    if "requestBody" in operation:
        body_content = operation["requestBody"].get("content", {})
        if "application/json" in body_content:
            body_schema = body_content["application/json"].get("schema", {})
            if body_schema.get("type") == "object":
                for prop_name, prop_schema in body_schema.get("properties", {}).items():
                    properties[prop_name] = openapi_to_json_schema(prop_schema)
                required.extend(body_schema.get("required", []))

    input_schema = {
        "type": "object",
        "properties": properties,
    }
    if required:
        input_schema["required"] = required

    return {
        "name": name,
        "description": description,
        "input_schema": input_schema,
    }

Фреймворк решений

Автоматически генерируйте из OpenAPI когда:

• API имеет 10+ endpoints, которые нужно экспозировать
• API спецификация активно поддерживается и часто обновляется
• Вам нужен read-only доступ к хорошо структурированным REST API
• Endpoint схемы полные с чёткими описаниями параметров
• Time-to-market критично (часы vs. дни на ручное создание)

Создавайте ручные инструменты когда:

• API не имеет OpenAPI спецификации или спеца устарела/неполная
• Вам нужны только 2-5 конкретных операций из большого API surface
• Требуется кастомная логика: трансформация ответа, многошаговые рабочие процессы, stateful операции
• Обработка ошибок требует domain-specific стратегий восстановления
• Инструмент требует комбинирования нескольких API вызовов в одну логическую операцию

Гибридный подход: автоматически генерируйте базовые инструменты из спеца, затем оборачивайте high-value операции в ручные инструменты, которые добавляют бизнес-логику.

Таблица параметров

Типичные ошибки

Ошибка 1: Отсутствует $ref разрешение

Влияние: генерация инструмента падает с KeyError или неполными схемами

❌ Неправильно:

# Directly using schema with $ref without resolution
param_schema = param["schema"]  # {"$ref": "#/components/schemas/User"}
properties[param_name] = param_schema  # Breaks tool validation

✅ Правильно:

def resolve_ref(spec: dict, ref: str) -> dict:
    """Resolve $ref pointer in OpenAPI spec."""
    parts = ref.lstrip("#/").split("/")
    current = spec
    for part in parts:
        current = current[part]
    return current

param_schema = param.get("schema", {})
if "$ref" in param_schema:
    param_schema = resolve_ref(spec, param_schema["$ref"])
properties[param_name] = openapi_to_json_schema(param_schema)

Ошибка 2: Path параметры не извлечены из входа инструмента

Влияние: ошибки 404 потому что path template {userId} не подставлено с реальным значением

❌ Неправильно:

# Sending path template directly
url = f"{base_url}/users/{{userId}}"  # Literal {userId} in URL
response = httpx.get(url)  # 404 Not Found

✅ Правильно:

def resolve_path_params(path: str, params: dict) -> tuple[str, dict]:
    """Extract path params and build final URL path."""
    import re
    path_params = re.findall(r'\{(\w+)\}', path)
    remaining = dict(params)

    for param in path_params:
        if param in remaining:
            value = remaining.pop(param)
            path = path.replace(f"{{{param}}}", str(value))

    return path, remaining

# Usage
path_template = "/users/{userId}/orders/{orderId}"
tool_input = {"userId": "123", "orderId": "456", "status": "shipped"}
final_path, query_params = resolve_path_params(path_template, tool_input)
# final_path = "/users/123/orders/456"
# query_params = {"status": "shipped"}

Ошибка 3: Игнорирование HTTP 429 Rate Limit ответов

Влияние: быстрые отказы каскадят; IP может быть забанена

❌ Неправильно:

async def execute(tool_name: str, tool_input: dict):
    response = await client.post(url, json=tool_input)
    return response.json()  # Crashes on 429 with JSONDecodeError

✅ Правильно:

async def execute_with_retry(tool_name: str, tool_input: dict, max_retries: int = 3):
    for attempt in range(max_retries):
        response = await client.post(url, json=tool_input)

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 10))
            await asyncio.sleep(retry_after)
            continue  # Retry after waiting

        if response.is_success:
            return response.json()
        else:
            return {"error": response.text, "status": response.status_code}

    return {"error": "Max retries exceeded due to rate limiting"}

Ошибка 4: Описания инструментов слишком расплывчатые

Влияние: агент использует инструменты неправильно или пропускает релевантные инструменты

❌ Неправильно:

"description": "Create customer"  # When? What fields required? What's returned?

✅ Правильно:

"description": "Create a new customer in Stripe. Requires email (string). Optional: name, phone, metadata. Returns customer object with id, email, created timestamp. Use when onboarding a new user who needs payment processing."

Ошибка 5: Отсутствуют лимиты размера результата

Влияние: результат инструмента превышает лимит Anthropic 10K символов; сообщение отклонено

❌ Неправильно:

result = response.json()
return json.dumps(result)  # Could be 100KB+ for large lists

✅ Правильно:

result = response.json()
result_str = json.dumps(result, default=str)

if len(result_str) > 10000:
    result_str = result_str[:9900] + "... [truncated]"

return result_str

OpenAPI → Tool преобразование

Полный workflow: загрузка spec → фильтрация endpoints → инъекция аутентификации → регистрация с executor

class OpenAPIToolGenerator:
    def __init__(
        self,
        spec: dict,
        include_tags: list[str] | None = None,
        exclude_tags: list[str] | None = None,
        max_tools: int = 50,
        read_only: bool = False
    ):
        self.spec = spec
        self.include_tags = include_tags
        self.exclude_tags = exclude_tags or []
        self.max_tools = max_tools
        self.read_only = read_only
        self.base_url = self._extract_base_url()

    def _extract_base_url(self) -> str:
        """Extract base URL from spec servers section."""
        if "servers" in self.spec and self.spec["servers"]:
            return self.spec["servers"][0]["url"].rstrip("/")
        return ""

    def _should_include(self, operation: dict) -> bool:
        """Check if operation matches tag filters."""
        tags = operation.get("tags", [])

        if self.include_tags:
            if not any(t in self.include_tags for t in tags):
                return False

        if self.exclude_tags:
            if any(t in self.exclude_tags for t in tags):
                return False

        return True

    def generate_tools(self) -> list[dict]:
        """Generate all tool definitions from spec."""
        tools = []
        paths = self.spec.get("paths", {})

        for path, path_item in paths.items():
            for method in ["get", "post", "put", "patch", "delete"]:
                if method not in path_item:
                    continue

                operation = path_item[method]

                if self.read_only and method in ["post", "put", "patch", "delete"]:
                    continue

                if not self._should_include(operation):
                    continue

                tool = generate_tool_from_endpoint(method, path, operation)
                if tool:
                    tools.append(tool)

                    if len(tools) >= self.max_tools:
                        return tools

        return tools

Инъекция аутентификации:

class APIToolExecutor:
    def __init__(self, base_url: str, auth_config: dict):
        self.base_url = base_url.rstrip("/")
        self.auth_config = auth_config

    def _build_auth_headers(self) -> dict:
        """Build authentication headers from config."""
        auth_type = self.auth_config.get("type", "none")

        if auth_type == "bearer":
            return {"Authorization": f"Bearer {self.auth_config['token']}"}
        elif auth_type == "api_key":
            key_name = self.auth_config.get("header", "X-API-Key")
            return {key_name: self.auth_config["key"]}
        elif auth_type == "basic":
            import base64
            creds = f"{self.auth_config['username']}:{self.auth_config['password']}"
            encoded = base64.b64encode(creds.encode()).decode()
            return {"Authorization": f"Basic {encoded}"}

        return {}

Ручные инструменты

Когда автоматическая генерация недостаточна, наследуйте BaseTool или используйте декоратор @tool для кастомной логики.

Использование декоратора @tool

from langchain.tools import tool
from pydantic import BaseModel, Field

class StripeRefundInput(BaseModel):
    charge_id: str = Field(description="Stripe charge ID (starts with 'ch_')")
    amount: int | None = Field(None, description="Amount to refund in cents. If None, refunds full charge.")
    reason: str = Field("requested_by_customer", description="Reason: duplicate, fraudulent, requested_by_customer")

@tool("stripe_refund_charge", args_schema=StripeRefundInput)
async def refund_charge(charge_id: str, amount: int | None = None, reason: str = "requested_by_customer") -> str:
    """Refund a Stripe charge. Use when customer requests refund or dispute occurs."""
    import stripe

    try:
        refund = stripe.Refund.create(
            charge=charge_id,
            amount=amount,
            reason=reason
        )
        return json.dumps({
            "success": True,
            "refund_id": refund.id,
            "amount": refund.amount,
            "status": refund.status
        })
    except stripe.error.InvalidRequestError as e:
        return json.dumps({"success": False, "error": str(e)})

Async паттерн инструмента для I/O-bound операций

from langchain.tools import BaseTool

class GitHubSearchIssues(BaseTool):
    name = "github_search_issues"
    description = "Search GitHub issues across repositories. Returns top 10 matching issues with title, body excerpt, URL."

    async def _arun(self, query: str, repo: str | None = None) -> str:
        """Async implementation using httpx."""
        import httpx

        search_query = f"{query} type:issue"
        if repo:
            search_query += f" repo:{repo}"

        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.github.com/search/issues",
                params={"q": search_query, "per_page": 10},
                headers={"Authorization": f"Bearer {self.github_token}"}
            )

            if response.status_code != 200:
                return json.dumps({"error": f"GitHub API returned {response.status_code}"})

            results = response.json()
            issues = [
                {
                    "title": item["title"],
                    "url": item["html_url"],
                    "state": item["state"],
                    "created": item["created_at"]
                }
                for item in results.get("items", [])[:10]
            ]

            return json.dumps({"total_count": results.get("total_count", 0), "issues": issues})

    def _run(self, query: str, repo: str | None = None) -> str:
        """Sync fallback."""
        import asyncio
        return asyncio.run(self._arun(query, repo))

Паттерн реестра инструментов

Динамическая загрузка инструментов с версионированием и A/B testing возможностями.

import hashlib
from datetime import datetime

class DynamicToolRegistry:
    def __init__(self):
        self.tools = {}           # tool_name -> tool_definition
        self.executors = {}       # api_name -> APIToolExecutor
        self.versions = {}        # api_name -> version hash
        self.metrics = {}         # tool_name -> {"calls": int, "errors": int, "avg_latency_ms": float}

    def register_api(
        self,
        name: str,
        spec_url: str,
        auth_config: dict,
        version: str | None = None,
        **generator_kwargs
    ) -> dict:
        """Register API and return registration metadata."""
        spec = load_openapi_spec(spec_url)

        # Version tracking
        spec_hash = hashlib.sha256(json.dumps(spec, sort_keys=True).encode()).hexdigest()[:8]
        version = version or spec_hash

        generator = OpenAPIToolGenerator(spec, **generator_kwargs)
        tools = generator.generate_tools()

        executor = APIToolExecutor(
            base_url=generator.base_url,
            auth_config=auth_config
        )

        # Register tools with API prefix
        for tool in tools:
            prefixed_name = f"{name}__{tool['name']}"
            self.tools[prefixed_name] = {
                **tool,
                "name": prefixed_name,
                "_api": name,
                "_version": version,
                "_registered_at": datetime.utcnow().isoformat()
            }
            self.metrics[prefixed_name] = {"calls": 0, "errors": 0, "latency_sum_ms": 0}

        self.executors[name] = executor
        self.versions[name] = version

        return {
            "api_name": name,
            "version": version,
            "tools_registered": len(tools),
            "base_url": generator.base_url
        }

    def get_tools_for_claude(self, api_filter: list[str] | None = None) -> list[dict]:
        """Get clean tool definitions for Claude. Optionally filter by API names."""
        result = []
        for tool_name, tool in self.tools.items():
            if api_filter and tool["_api"] not in api_filter:
                continue

            # Remove internal metadata
            clean = {k: v for k, v in tool.items() if not k.startswith("_")}
            result.append(clean)

        return result

    async def execute_tool(self, tool_name: str, tool_input: dict) -> str:
        """Execute tool and track metrics."""
        import time

        start = time.monotonic()

        if tool_name not in self.tools:
            return json.dumps({"error": f"Tool not found: {tool_name}"})

        api_name = self.tools[tool_name]["_api"]
        executor = self.executors[api_name]

        original_name = tool_name.replace(f"{api_name}__", "", 1)

        try:
            result = await executor.execute(original_name, tool_input)

            # Update metrics
            latency_ms = (time.monotonic() - start) * 1000
            self.metrics[tool_name]["calls"] += 1
            self.metrics[tool_name]["latency_sum_ms"] += latency_ms

            if '"error"' in result or '"_error": true' in result:
                self.metrics[tool_name]["errors"] += 1

            return result
        except Exception as e:
            self.metrics[tool_name]["calls"] += 1
            self.metrics[tool_name]["errors"] += 1
            return json.dumps({"error": str(e)})

    def get_metrics(self, tool_name: str | None = None) -> dict:
        """Get performance metrics for tools."""
        if tool_name:
            m = self.metrics.get(tool_name, {})
            if m["calls"] > 0:
                m["avg_latency_ms"] = m["latency_sum_ms"] / m["calls"]
                m["error_rate"] = m["errors"] / m["calls"]
            return m

        # Return all metrics
        return {
            name: {
                **m,
                "avg_latency_ms": m["latency_sum_ms"] / m["calls"] if m["calls"] > 0 else 0,
                "error_rate": m["errors"] / m["calls"] if m["calls"] > 0 else 0
            }
            for name, m in self.metrics.items()
        }

A/B testing версий инструментов:

def register_api_ab_test(
    registry: DynamicToolRegistry,
    name: str,
    spec_url_a: str,
    spec_url_b: str,
    auth_config: dict,
    traffic_split: float = 0.5  # 0.0-1.0, fraction going to version A
):
    """Register two versions of an API for A/B testing."""
    registry.register_api(f"{name}_v1", spec_url_a, auth_config)
    registry.register_api(f"{name}_v2", spec_url_b, auth_config)

    # Route based on traffic split
    def get_tools_ab():
        import random
        if random.random() < traffic_split:
            return registry.get_tools_for_claude(api_filter=[f"{name}_v1"])
        else:
            return registry.get_tools_for_claude(api_filter=[f"{name}_v2"])

    return get_tools_ab

Производительность и бенчмарки

Примечание: Приведённые ниже цифры являются иллюстративными оценками на основе типичных production-конфигураций, а не измерениями конкретной системы.

Скорость генерации инструментов:

• Маленький API (10 endpoints): обычно завершается менее чем за 100ms
• Средний API (50 endpoints, похожий на Stripe): обычно 200-400ms
• Большой API (200+ endpoints, похожий на AWS): обычно 1-2 секунды с полным разрешением $ref

Overhead выполнения во время выполнения:

• Маршрутизация вызова инструмента через реестр: добавляет приблизительно 1-3ms за вызов
• Построение заголовков аутентификации: пренебрежимо (менее 0.5ms)
• Проверка rate limiting (token bucket): обычно менее 1ms
• Полная latency выполнения инструмента: доминируется network round-trip (50-500ms типично для REST API)

Влияние окна контекста:

• Каждое определение инструмента: обычно 150-400 токенов в зависимости от сложности параметров
• 50 инструментов: приблизительно 10,000-20,000 токенов контекста
• С Claude 3.5 Sonnet 200K контекстом: может вместить сотни инструментов, хотя производительность деградирует с слишком большим количеством вариантов

Стратегии оптимизации:

• Фильтрация на основе tags значительно уменьшает количество инструментов (часто на 60-80% для больших API)
• Lazy loading: регистрируйте API только когда агент сигнализирует намерение их использовать
• Консолидация инструментов: объедините похожие операции (напр., get_user_by_id + get_user_by_email → get_user с гибким входом)

Паттерны успеха восстановления ошибок:

• С retry + exponential backoff: переходящие ошибки (429, 503) обычно разрешаются в 2-3 попыток
• С структурированными ошибками ответов: агент успешно восстанавливается от ошибок валидации параметров в последующих ходах
• Без контекста ошибки: агент часто повторяет ту же ошибку или сдаётся преждевременно