FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
python-sdk/examples/echo_agent.py at package-scripts · openSUSE-Python/python-sdk · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
openSUSE-Python
/
python-sdk
Public
forked from
agentclientprotocol/python-sdk
Notifications
You must be signed in to change notification settings
Fork
0
Star
0
Code
Pull requests
0
Actions
Projects
Security and quality
0
Insights
Additional navigation options
Code
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
python-sdk
/
examples
/
echo_agent.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
86 lines (74 loc) · 2.18 KB
Breadcrumbs
python-sdk
/
examples
/
echo_agent.py
Copy path
File metadata and controls
86 lines (74 loc) · 2.18 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
# /// script
# requires-python = ">=3.10,<3.15"
# dependencies = [
# "agent-client-protocol",
# ]
# ///
import
asyncio
from
typing
import
Any
from
uuid
import
uuid4
from
acp
import
(
Agent
,
InitializeResponse
,
NewSessionResponse
,
PromptResponse
,
run_agent
,
text_block
,
update_agent_message
,
)
from
acp
.
interfaces
import
Client
from
acp
.
schema
import
(
AudioContentBlock
,
ClientCapabilities
,
EmbeddedResourceContentBlock
,
HttpMcpServer
,
ImageContentBlock
,
Implementation
,
McpServerStdio
,
ResourceContentBlock
,
SseMcpServer
,
TextContentBlock
,
)
class
EchoAgent
(
Agent
):
_conn
:
Client
def
on_connect
(
self
,
conn
:
Client
)
->
None
:
self
.
_conn
=
conn
async
def
initialize
(
self
,
protocol_version
:
int
,
client_capabilities
:
ClientCapabilities
|
None
=
None
,
client_info
:
Implementation
|
None
=
None
,
**
kwargs
:
Any
,
)
->
InitializeResponse
:
return
InitializeResponse
(
protocol_version
=
protocol_version
)
async
def
new_session
(
self
,
cwd
:
str
,
additional_directories
:
list
[
str
]
|
None
=
None
,
mcp_servers
:
list
[
HttpMcpServer
|
SseMcpServer
|
McpServerStdio
]
|
None
=
None
,
**
kwargs
:
Any
,
)
->
NewSessionResponse
:
return
NewSessionResponse
(
session_id
=
uuid4
().
hex
)
async
def
prompt
(
self
,
session_id
:
str
,
prompt
:
list
[
TextContentBlock
|
ImageContentBlock
|
AudioContentBlock
|
ResourceContentBlock
|
EmbeddedResourceContentBlock
],
**
kwargs
:
Any
,
)
->
PromptResponse
:
for
block
in
prompt
:
text
=
block
.
get
(
"text"
,
""
)
if
isinstance
(
block
,
dict
)
else
getattr
(
block
,
"text"
,
""
)
chunk
=
update_agent_message
(
text_block
(
text
))
chunk
.
field_meta
=
{
"echo"
:
True
}
chunk
.
content
.
field_meta
=
{
"echo"
:
True
}
await
self
.
_conn
.
session_update
(
session_id
=
session_id
,
update
=
chunk
,
source
=
"echo_agent"
)
return
PromptResponse
(
stop_reason
=
"end_turn"
)
async
def
main
()
->
None
:
await
run_agent
(
EchoAgent
())
if
__name__
==
"__main__"
:
asyncio
.
run
(
main
())
Back
|
FazBrowse Home
|
New Git URL