68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
"""
|
|
Echo plugin for agentsd - verifies the gRPC syscall chain works.
|
|
Registers as a basic plugin, calls Display.Text and FS.Read.
|
|
"""
|
|
import grpc
|
|
import sys
|
|
import os
|
|
|
|
# Add SDK path
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "_sdk"))
|
|
|
|
from agentsd_pb2 import (
|
|
DisplayTextRequest, FsReadRequest, FsPathRequest,
|
|
MemoryWriteRequest, MemoryReadRequest,
|
|
PluginInfo, ExportDef, Empty
|
|
)
|
|
from agentsd_pb2_grpc import (
|
|
DisplayStub, FSStub, MemoryStub, PluginBridgeStub
|
|
)
|
|
|
|
|
|
PLUGIN_ID = "echo-test"
|
|
METADATA = (("x-plugin-id", PLUGIN_ID),)
|
|
|
|
|
|
def main():
|
|
channel = grpc.insecure_channel('localhost:50051')
|
|
|
|
# Register plugin
|
|
bridge = PluginBridgeStub(channel)
|
|
resp = bridge.Register(PluginInfo(
|
|
id=PLUGIN_ID,
|
|
name="Echo Test Plugin",
|
|
version="0.1.0",
|
|
plugin_type="basic",
|
|
syscalls=["display.text", "fs.list", "memory.read", "memory.write"],
|
|
depends=[],
|
|
exports=[ExportDef(name="echo", description="Echo test")]
|
|
))
|
|
print(f"Register: ok={resp.ok} session={resp.session_id}")
|
|
|
|
# Test Display
|
|
display = DisplayStub(channel)
|
|
resp = display.Text(DisplayTextRequest(content="Hello from echo plugin!"), metadata=METADATA)
|
|
print(f"Display.Text: ok={resp.ok}")
|
|
|
|
# Test Memory write + read
|
|
memory = MemoryStub(channel)
|
|
resp = memory.Write(MemoryWriteRequest(key="test.greeting", value=b"Hello World"), metadata=METADATA)
|
|
print(f"Memory.Write: ok={resp.ok}")
|
|
|
|
resp = memory.Read(MemoryReadRequest(key="test.greeting"), metadata=METADATA)
|
|
print(f"Memory.Read: found={resp.found} value={resp.value.decode()}")
|
|
|
|
# Test FS list
|
|
fs = FSStub(channel)
|
|
resp = fs.List(FsPathRequest(path="."), metadata=METADATA)
|
|
print(f"FS.List: ok={resp.ok} entries={len(resp.entries)}")
|
|
for entry in resp.entries[:5]:
|
|
print(f" {entry.name} {'[dir]' if entry.is_dir else ''}")
|
|
|
|
print("\nAll syscalls verified OK!")
|
|
channel.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|