forked from themanojdesai/python-a2a
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathconversion.py
More file actions
177 lines (147 loc) · 4.96 KB
/
Copy pathconversion.py
File metadata and controls
177 lines (147 loc) · 4.96 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
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
107
108
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
"""
Conversion utilities for working with A2A messages and conversations.
"""
import uuid
from typing import Dict, List, Any, Optional, Union
from ..models.message import Message, MessageRole
from ..models.content import (
TextContent, FunctionCallContent, FunctionResponseContent,
ErrorContent, FunctionParameter
)
from ..models.conversation import Conversation
def create_text_message(
text: str,
role: MessageRole = MessageRole.USER,
conversation_id: Optional[str] = None,
parent_message_id: Optional[str] = None,
message_id: Optional[str] = None
) -> Message:
"""
Create a simple text message
Args:
text: The text content of the message
role: The role of the sender (default: USER)
conversation_id: Optional conversation ID
parent_message_id: Optional parent message ID
message_id: Optional message ID (will be generated if not provided)
Returns:
A new Message object
"""
content = TextContent(text=text)
return Message(
content=content,
role=role,
message_id=message_id or str(uuid.uuid4()),
conversation_id=conversation_id,
parent_message_id=parent_message_id
)
def create_function_call(
function_name: str,
parameters: List[Dict[str, Any]],
role: MessageRole = MessageRole.USER,
conversation_id: Optional[str] = None,
parent_message_id: Optional[str] = None,
message_id: Optional[str] = None
) -> Message:
"""
Create a function call message
Args:
function_name: The name of the function to call
parameters: List of parameter dictionaries with 'name' and 'value' keys
role: The role of the sender (default: USER)
conversation_id: Optional conversation ID
parent_message_id: Optional parent message ID
message_id: Optional message ID (will be generated if not provided)
Returns:
A new Message object
"""
function_params = [
FunctionParameter(name=param["name"], value=param["value"])
for param in parameters
]
content = FunctionCallContent(
name=function_name,
parameters=function_params
)
return Message(
content=content,
role=role,
message_id=message_id or str(uuid.uuid4()),
conversation_id=conversation_id,
parent_message_id=parent_message_id
)
def create_function_response(
function_name: str,
response: Any,
role: MessageRole = MessageRole.AGENT,
conversation_id: Optional[str] = None,
parent_message_id: Optional[str] = None,
message_id: Optional[str] = None
) -> Message:
"""
Create a function response message
Args:
function_name: The name of the function that was called
response: The response data
role: The role of the sender (default: AGENT)
conversation_id: Optional conversation ID
parent_message_id: Optional parent message ID
message_id: Optional message ID (will be generated if not provided)
Returns:
A new Message object
"""
content = FunctionResponseContent(
name=function_name,
response=response
)
return Message(
content=content,
role=role,
message_id=message_id or str(uuid.uuid4()),
conversation_id=conversation_id,
parent_message_id=parent_message_id
)
def create_error_message(
error_message: str,
role: MessageRole = MessageRole.SYSTEM,
conversation_id: Optional[str] = None,
parent_message_id: Optional[str] = None,
message_id: Optional[str] = None
) -> Message:
"""
Create an error message
Args:
error_message: The error message text
role: The role of the sender (default: SYSTEM)
conversation_id: Optional conversation ID
parent_message_id: Optional parent message ID
message_id: Optional message ID (will be generated if not provided)
Returns:
A new Message object
"""
content = ErrorContent(message=error_message)
return Message(
content=content,
role=role,
message_id=message_id or str(uuid.uuid4()),
conversation_id=conversation_id,
parent_message_id=parent_message_id
)
def format_function_params(params: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Convert a dictionary of parameters to the format expected by function calls
Args:
params: Dictionary of parameter name-value pairs
Returns:
List of parameter dictionaries with 'name' and 'value' keys
"""
return [{"name": key, "value": value} for key, value in params.items()]
def conversation_to_messages(conversation: Conversation) -> List[Message]:
"""
Extract messages from a conversation
Args:
conversation: The conversation to extract messages from
Returns:
List of messages from the conversation
"""
return conversation.messages