# Copyright AGNTCY Contributors (https://github.com/agntcy)
# SPDX-License-Identifier: Apache-2.0
# coding: utf-8
"""
Agent Connect Protocol
No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)
The version of the OpenAPI document: 0.2.2
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
from __future__ import annotations
import json
import pprint
import re # noqa: F401
from typing import Any, ClassVar, Dict, List, Optional, Set
from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr, field_validator
from typing_extensions import Annotated, Self
from agntcy_acp.acp_v0.models.config import Config
from agntcy_acp.acp_v0.models.stream_mode import StreamMode
[docs]
class RunCreateStateless(BaseModel):
"""
Payload for creating a stateless run.
""" # noqa: E501
agent_id: Optional[StrictStr] = Field(
default=None,
description="The agent ID to run. If not provided will use the default agent for this service.",
)
input: Optional[Dict[str, Any]] = Field(
default=None,
description="The input to the agent. The schema is described in agent ACP descriptor under 'spec.thread_state'.'input'.",
)
metadata: Optional[Dict[str, Any]] = Field(
default=None, description="Metadata to assign to the run."
)
config: Optional[Config] = None
webhook: Optional[
Annotated[str, Field(min_length=1, strict=True, max_length=65536)]
] = Field(
default=None,
description="Webhook to call upon change of run status. This is a url that accepts a POST containing the `Run` object as body. See Callbacks definition.",
)
stream_mode: Optional[StreamMode] = None
on_disconnect: Optional[StrictStr] = Field(
default="cancel",
description="The disconnect mode to use. Must be one of 'cancel' or 'continue'.",
)
multitask_strategy: Optional[StrictStr] = Field(
default="reject",
description="Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'.",
)
after_seconds: Optional[StrictInt] = Field(
default=None,
description="The number of seconds to wait before starting the run. Use to schedule future runs.",
)
on_completion: Optional[StrictStr] = Field(
default="delete",
description="Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'.",
)
__properties: ClassVar[List[str]] = [
"agent_id",
"input",
"metadata",
"config",
"webhook",
"stream_mode",
"on_disconnect",
"multitask_strategy",
"after_seconds",
"on_completion",
]
[docs]
@field_validator("on_disconnect")
def on_disconnect_validate_enum(cls, value):
"""Validates the enum"""
if value is None:
return value
if value not in set(["cancel", "continue"]):
raise ValueError("must be one of enum values ('cancel', 'continue')")
return value
[docs]
@field_validator("multitask_strategy")
def multitask_strategy_validate_enum(cls, value):
"""Validates the enum"""
if value is None:
return value
if value not in set(["reject", "rollback", "interrupt", "enqueue"]):
raise ValueError(
"must be one of enum values ('reject', 'rollback', 'interrupt', 'enqueue')"
)
return value
[docs]
@field_validator("on_completion")
def on_completion_validate_enum(cls, value):
"""Validates the enum"""
if value is None:
return value
if value not in set(["delete", "keep"]):
raise ValueError("must be one of enum values ('delete', 'keep')")
return value
model_config = ConfigDict(
populate_by_name=True,
validate_assignment=True,
protected_namespaces=(),
)
[docs]
def to_str(self) -> str:
"""Returns the string representation of the model using alias"""
return pprint.pformat(self.model_dump(by_alias=True))
[docs]
def to_json(self) -> str:
"""Returns the JSON representation of the model using alias"""
# TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
return self.model_dump_json(by_alias=True, exclude_unset=True)
[docs]
@classmethod
def from_json(cls, json_str: str) -> Optional[Self]:
"""Create an instance of RunCreateStateless from a JSON string"""
return cls.from_dict(json.loads(json_str))
[docs]
def to_dict(self) -> Dict[str, Any]:
"""Return the dictionary representation of the model using alias.
This has the following differences from calling pydantic's
`self.model_dump(by_alias=True)`:
* `None` is only added to the output dict for nullable fields that
were set at model initialization. Other fields with value `None`
are ignored.
"""
excluded_fields: Set[str] = set([])
_dict = self.model_dump(
by_alias=True,
exclude=excluded_fields,
exclude_none=True,
)
# override the default output from pydantic by calling `to_dict()` of config
if self.config:
_dict["config"] = self.config.to_dict()
# override the default output from pydantic by calling `to_dict()` of stream_mode
if self.stream_mode:
_dict["stream_mode"] = self.stream_mode.to_dict()
# set to None if stream_mode (nullable) is None
# and model_fields_set contains the field
if self.stream_mode is None and "stream_mode" in self.model_fields_set:
_dict["stream_mode"] = None
return _dict
[docs]
@classmethod
def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
"""Create an instance of RunCreateStateless from a dict"""
if obj is None:
return None
if not isinstance(obj, dict):
return cls.model_validate(obj)
_obj = cls.model_validate(
{
"agent_id": obj.get("agent_id"),
"input": obj.get("input"),
"metadata": obj.get("metadata"),
"config": Config.from_dict(obj["config"])
if obj.get("config") is not None
else None,
"webhook": obj.get("webhook"),
"stream_mode": StreamMode.from_dict(obj["stream_mode"])
if obj.get("stream_mode") is not None
else None,
"on_disconnect": obj.get("on_disconnect")
if obj.get("on_disconnect") is not None
else "cancel",
"multitask_strategy": obj.get("multitask_strategy")
if obj.get("multitask_strategy") is not None
else "reject",
"after_seconds": obj.get("after_seconds"),
"on_completion": obj.get("on_completion")
if obj.get("on_completion") is not None
else "delete",
}
)
return _obj
