Skip to content

protocol

Protocol definitions for agent interfaces in the agenty framework.

This module defines the core protocols that all agents must implement. It provides two main protocols:

  • AgentIOProtocol: Defines the basic input/output interface for agents
  • AgentProtocol: Extends AgentIOProtocol with additional agent-specific functionality

These protocols ensure consistent behavior across different agent implementations and enable composition of agents in pipelines.

AgentIOProtocol

Bases: Generic[AgentInputT, AgentOutputT], Protocol

Protocol defining the basic input/output interface for agents.

This protocol establishes the core contract that agents must fulfill to process input data and produce output. It supports both synchronous and asynchronous execution, and enables pipeline composition through the | operator.

Type Parameters

AgentInputT: The type of input data the agent accepts AgentOutputT: The type of output data the agent produces

Source code in agenty/protocol.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class AgentIOProtocol(Generic[AgentInputT, AgentOutputT], Protocol):
    """Protocol defining the basic input/output interface for agents.

    This protocol establishes the core contract that agents must fulfill to process
    input data and produce output. It supports both synchronous and asynchronous
    execution, and enables pipeline composition through the | operator.

    Type Parameters:
        AgentInputT: The type of input data the agent accepts
        AgentOutputT: The type of output data the agent produces
    """

    @property
    def input_schema(self) -> Type[AgentIO]:
        """Get the input schema type for this agent.

        Returns:
            Type[AgentIO]: The Pydantic model class defining the expected input structure
        """
        ...

    @property
    def output_schema(self) -> Type[AgentIO]:
        """Get the output schema type for this agent.

        Returns:
            Type[AgentIO]: The Pydantic model class defining the produced output structure
        """
        ...

    async def run(
        self,
        input_data: Optional[AgentInputT],
        name: Optional[str] = None,
    ) -> AgentOutputT:
        """Execute the agent asynchronously with the given input data.

        Args:
            input_data: The input data to process, must conform to input_schema
            name: Optional name for this execution context

        Returns:
            The processed output data conforming to output_schema
        """
        ...

    def run_sync(
        self,
        input_data: Optional[AgentInputT],
        name: Optional[str] = None,
    ) -> AgentOutputT:
        """Execute the agent synchronously with the given input data.

        Args:
            input_data: The input data to process, must conform to input_schema
            name: Optional name for this execution context

        Returns:
            The processed output data conforming to output_schema
        """
        ...

    def reset(self) -> None:
        """Reset the agent's internal state.

        This method should clear any cached data or state that could affect
        future executions.
        """
        ...

    def __or__(
        self,
        other: "AgentIOProtocol[AgentOutputT, PipelineOutputT]",
    ) -> "AgentIOProtocol[AgentInputT, PipelineOutputT]":
        """Compose this agent with another agent to form a pipeline.

        This operator enables functional composition of agents, where the output
        of this agent becomes the input to the other agent.

        Args:
            other: Another agent whose input type matches this agent's output type

        Returns:
            A new agent representing the composed pipeline
        """
        ...

input_schema property

Get the input schema type for this agent.

Returns:

Type Description
Type[AgentIO]

Type[AgentIO]: The Pydantic model class defining the expected input structure

output_schema property

Get the output schema type for this agent.

Returns:

Type Description
Type[AgentIO]

Type[AgentIO]: The Pydantic model class defining the produced output structure

reset()

Reset the agent's internal state.

This method should clear any cached data or state that could affect future executions.

Source code in agenty/protocol.py 
83
84
85
86
87
88
89
def reset(self) -> None:
    """Reset the agent's internal state.

    This method should clear any cached data or state that could affect
    future executions.
    """
    ...

run(input_data, name=None) async

Execute the agent asynchronously with the given input data.

Parameters:

Name Type Description Default
input_data Optional[AgentInputT]

The input data to process, must conform to input_schema

 required
name Optional[str]

Optional name for this execution context

 None

Returns:

Type Description
AgentOutputT

The processed output data conforming to output_schema
Source code in agenty/protocol.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
async def run(
    self,
    input_data: Optional[AgentInputT],
    name: Optional[str] = None,
) -> AgentOutputT:
    """Execute the agent asynchronously with the given input data.

    Args:
        input_data: The input data to process, must conform to input_schema
        name: Optional name for this execution context

    Returns:
        The processed output data conforming to output_schema
    """
    ...

run_sync(input_data, name=None)

Execute the agent synchronously with the given input data.

Parameters:

Name Type Description Default
input_data Optional[AgentInputT]

The input data to process, must conform to input_schema

 required
name Optional[str]

Optional name for this execution context

 None

Returns:

Type Description
AgentOutputT

The processed output data conforming to output_schema
Source code in agenty/protocol.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def run_sync(
    self,
    input_data: Optional[AgentInputT],
    name: Optional[str] = None,
) -> AgentOutputT:
    """Execute the agent synchronously with the given input data.

    Args:
        input_data: The input data to process, must conform to input_schema
        name: Optional name for this execution context

    Returns:
        The processed output data conforming to output_schema
    """
    ...

AgentProtocol

Bases: AgentIOProtocol[AgentInputT, AgentOutputT], Protocol

Protocol extending AgentIOProtocol with agent-specific functionality.

This protocol adds properties for agent identification and chat history management on top of the basic input/output operations defined in AgentIOProtocol.

Type Parameters

AgentInputT: The type of input data the agent accepts AgentOutputT: The type of output data the agent produces

Source code in agenty/protocol.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class AgentProtocol(AgentIOProtocol[AgentInputT, AgentOutputT], Protocol):
    """Protocol extending AgentIOProtocol with agent-specific functionality.

    This protocol adds properties for agent identification and chat history
    management on top of the basic input/output operations defined in AgentIOProtocol.

    Type Parameters:
        AgentInputT: The type of input data the agent accepts
        AgentOutputT: The type of output data the agent produces
    """

    @property
    def name(self) -> str:
        """Get the name of this agent instance.

        Returns:
            str: The unique identifier for this agent
        """
        ...

    @property
    def chat_history(self) -> ChatHistory:
        """Get the chat history for this agent.

        Returns:
            ChatHistory: The object tracking all interactions with this agent
        """
        ...

    @chat_history.setter
    def chat_history(self, value: ChatHistory) -> None:
        """Set the chat history for this agent.

        Args:
            value: The new chat history object to use
        """
        ...

chat_history property writable

Get the chat history for this agent.

Returns:

Name Type Description
ChatHistory ChatHistory

The object tracking all interactions with this agent

name property

Get the name of this agent instance.

Returns:

Name Type Description
str str

The unique identifier for this agent