python-sdk/examples/client.py at package-scripts · openSUSE-Python/python-sdk · GitHub

History

229 lines (191 loc) · 6.9 KB

Raw

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

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

import asyncio

import asyncio.subprocess as aio_subprocess

import contextlib

import logging

import os

import sys

from pathlib import Path

from typing import Any

from acp import (

PROTOCOL_VERSION,

Client,

RequestError,

connect_to_agent,

text_block,

)

from acp.core import ClientSideConnection

from acp.schema import (

AgentMessageChunk,

AgentPlanContentUpdate,

AgentPlanRemovedUpdate,

AgentPlanUpdate,

AgentThoughtChunk,

AudioContentBlock,

AvailableCommandsUpdate,

ClientCapabilities,

ConfigOptionUpdate,

CreateElicitationResponse,

CreateTerminalResponse,

CurrentModeUpdate,

DeclineElicitationResponse,

ElicitationMode,

EmbeddedResourceContentBlock,

EnvVariable,

ImageContentBlock,

Implementation,

KillTerminalResponse,

PermissionOption,

ReadTextFileResponse,

ReleaseTerminalResponse,

RequestPermissionResponse,

ResourceContentBlock,

SessionInfoUpdate,

TerminalOutputResponse,

TextContentBlock,

ToolCallProgress,

ToolCallStart,

ToolCallUpdate,

UsageUpdate,

UserMessageChunk,

WaitForTerminalExitResponse,

WriteTextFileResponse,

)

class ExampleClient(Client):

async def request_permission(

self, session_id: str, tool_call: ToolCallUpdate, options: list[PermissionOption], **kwargs: Any

) -> RequestPermissionResponse:

raise RequestError.method_not_found("session/request_permission")

async def write_text_file(

self, session_id: str, path: str, content: str, **kwargs: Any

) -> WriteTextFileResponse | None:

raise RequestError.method_not_found("fs/write_text_file")

async def read_text_file(

self, session_id: str, path: str, line: int | None = None, limit: int | None = None, **kwargs: Any

) -> ReadTextFileResponse:

raise RequestError.method_not_found("fs/read_text_file")

async def create_terminal(

self,

session_id: str,

command: str,

args: list[str] | None = None,

env: list[EnvVariable] | None = None,

cwd: str | None = None,

output_byte_limit: int | None = None,

**kwargs: Any,

) -> CreateTerminalResponse:

raise RequestError.method_not_found("terminal/create")

async def terminal_output(self, session_id: str, terminal_id: str, **kwargs: Any) -> TerminalOutputResponse:

raise RequestError.method_not_found("terminal/output")

async def release_terminal(

self, session_id: str, terminal_id: str, **kwargs: Any

) -> ReleaseTerminalResponse | None:

raise RequestError.method_not_found("terminal/release")

async def wait_for_terminal_exit(

self, session_id: str, terminal_id: str, **kwargs: Any

) -> WaitForTerminalExitResponse:

raise RequestError.method_not_found("terminal/wait_for_exit")

async def kill_terminal(self, session_id: str, terminal_id: str, **kwargs: Any) -> KillTerminalResponse | None:

raise RequestError.method_not_found("terminal/kill")

async def create_elicitation(self, message: str, mode: ElicitationMode, **kwargs: Any) -> CreateElicitationResponse:

print(f"| Agent requested input: {message} ({type(mode).__name__})")

return DeclineElicitationResponse(action="decline")

async def complete_elicitation(self, elicitation_id: str, **kwargs: Any) -> None:

print(f"| Agent completed elicitation: {elicitation_id}")

async def session_update(

self,

session_id: str,

update: UserMessageChunk

| AgentMessageChunk

| AgentThoughtChunk

| ToolCallStart

| ToolCallProgress

| AgentPlanUpdate

| AgentPlanContentUpdate

| AgentPlanRemovedUpdate

| AvailableCommandsUpdate

| CurrentModeUpdate

| ConfigOptionUpdate

| SessionInfoUpdate

| UsageUpdate,

**kwargs: Any,

) -> None:

if not isinstance(update, AgentMessageChunk):

return

content = update.content

text: str

if isinstance(content, TextContentBlock):

text = content.text

elif isinstance(content, ImageContentBlock):

text = "<image>"

elif isinstance(content, AudioContentBlock):

text = "<audio>"

elif isinstance(content, ResourceContentBlock):

text = content.uri or "<resource>"

elif isinstance(content, EmbeddedResourceContentBlock):

text = "<resource>"

else:

text = "<content>"

print(f"| Agent: {text}")

async def ext_method(self, method: str, params: dict) -> dict:

raise RequestError.method_not_found(method)

async def ext_notification(self, method: str, params: dict) -> None:

raise RequestError.method_not_found(method)

async def read_console(prompt: str) -> str:

loop = asyncio.get_running_loop()

return await loop.run_in_executor(None, lambda: input(prompt))

async def interactive_loop(conn: ClientSideConnection, session_id: str) -> None:

while True:

try:

line = await read_console("> ")

except EOFError:

break

except KeyboardInterrupt:

print("", file=sys.stderr)

break

if not line:

continue

try:

await conn.prompt(

session_id=session_id,

prompt=[text_block(line)],

)

except Exception as exc:

logging.error("Prompt failed: %s", exc) # noqa: TRY400

async def main(argv: list[str]) -> int:

logging.basicConfig(level=logging.INFO)

if len(argv) < 2:

print("Usage: python examples/client.py AGENT_PROGRAM [ARGS...]", file=sys.stderr)

return 2

program = argv[1]

args = argv[2:]

program_path = Path(program)

spawn_program = program

spawn_args = args

if program_path.exists() and not os.access(program_path, os.X_OK):

spawn_program = sys.executable

spawn_args = [str(program_path), *args]

proc = await asyncio.create_subprocess_exec(

spawn_program,

*spawn_args,

stdin=aio_subprocess.PIPE,

stdout=aio_subprocess.PIPE,

)

if proc.stdin is None or proc.stdout is None:

print("Agent process does not expose stdio pipes", file=sys.stderr)

return 1

client_impl = ExampleClient()

conn = connect_to_agent(client_impl, proc.stdin, proc.stdout)

await conn.initialize(

protocol_version=PROTOCOL_VERSION,

client_capabilities=ClientCapabilities(),

client_info=Implementation(name="example-client", title="Example Client", version="0.1.0"),

)

session = await conn.new_session(mcp_servers=[], cwd=os.getcwd())

await interactive_loop(conn, session.session_id)

if proc.returncode is None:

proc.terminate()

with contextlib.suppress(ProcessLookupError):

await proc.wait()

return 0

if __name__ == "__main__":

raise SystemExit(asyncio.run(main(sys.argv)))

python-sdk/examples/client.py at package-scripts · openSUSE-Python/python-sdk · GitHub

Uh oh!

Expand file tree

Breadcrumbs

client.py

Latest commit

History

Breadcrumbs

client.py

File metadata and controls