FilesExpand file tree

 main

/

echo_agent.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

82 lines (70 loc) · 2.08 KB

 main

/

echo_agent.py

Copy path

Top

File metadata and controls

• Code
• Blame

82 lines (70 loc) · 2.08 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

# /// 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, mcp_servers: list[HttpMcpServer | SseMcpServer | McpServerStdio], **kwargs: Any

) -> NewSessionResponse:

return NewSessionResponse(session_id=uuid4().hex)

async def prompt(

self,

prompt: list[

TextContentBlock

| ImageContentBlock

| AudioContentBlock

| ResourceContentBlock

| EmbeddedResourceContentBlock

],

session_id: str,

**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())