Source code for agntcy_acp.acp_v0.models.agent_capabilities
# 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, StrictBool
from typing_extensions import Self
from agntcy_acp.acp_v0.models.streaming_modes import StreamingModes
[docs]
class AgentCapabilities(BaseModel):
"""
Declares what invocation features this agent is capable of.
""" # noqa: E501
threads: Optional[StrictBool] = Field(
default=False,
description="This is `true` if the agent supports run threads. If this is `false`, then the threads tagged with `Threads` are not available. If missing, it means `false`",
)
interrupts: Optional[StrictBool] = Field(
default=False,
description="This is `true` if the agent runs can interrupt to request additional input and can be subsequently resumed. If missing, it means `false`",
)
callbacks: Optional[StrictBool] = Field(
default=False,
description="This is `true` if the agent supports a webhook to report run results. If this is `false`, providing a `webhook` at run creation has no effect. If missing, it means `false`",
)
streaming: Optional[StreamingModes] = None
__properties: ClassVar[List[str]] = [
"threads",
"interrupts",
"callbacks",
"streaming",
]
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 AgentCapabilities 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 streaming
if self.streaming:
_dict["streaming"] = self.streaming.to_dict()
return _dict
[docs]
@classmethod
def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
"""Create an instance of AgentCapabilities from a dict"""
if obj is None:
return None
if not isinstance(obj, dict):
return cls.model_validate(obj)
_obj = cls.model_validate(
{
"threads": obj.get("threads")
if obj.get("threads") is not None
else False,
"interrupts": obj.get("interrupts")
if obj.get("interrupts") is not None
else False,
"callbacks": obj.get("callbacks")
if obj.get("callbacks") is not None
else False,
"streaming": StreamingModes.from_dict(obj["streaming"])
if obj.get("streaming") is not None
else None,
}
)
return _obj