feat(hanzo-zap): add ZAP SDK package to monorepo (0.6.1)

Zero-copy Agent Protocol SDK — client, server, types.
Patch bump from PyPI 0.6.0 to bring into monorepo workspace.
This commit is contained in:
2026-03-12 18:30:18 -07:00
parent 2c7d718f6a
commit 748f342df3
7 changed files with 587 additions and 0 deletions
+38
View File
@@ -0,0 +1,38 @@
"""
hanzo-zap - Zero-copy Agent Protocol SDK for Python
1000x faster than MCP/JSON-RPC through binary wire protocol.
Example:
>>> from hanzo_zap import ZapClient
>>> async with ZapClient.connect("zap://localhost:9999") as client:
... tools = await client.list_tools()
... result = await client.call_tool("read_file", {"path": "README.md"})
"""
from .types import (
ApprovalPolicy,
SandboxPolicy,
MessageType,
Tool,
ToolCall,
ToolResult,
ServerInfo,
ClientInfo,
)
from .client import ZapClient
from .server import ZapServer
__version__ = "0.6.1"
__all__ = [
"ZapClient",
"ZapServer",
"ApprovalPolicy",
"SandboxPolicy",
"MessageType",
"Tool",
"ToolCall",
"ToolResult",
"ServerInfo",
"ClientInfo",
]
+171
View File
@@ -0,0 +1,171 @@
"""
ZAP Client implementation.
"""
from __future__ import annotations
import asyncio
import json
import ssl
import struct
from typing import Any
from urllib.parse import urlparse
from .types import (
ClientInfo,
MessageType,
Resource,
ServerInfo,
Tool,
ToolCall,
ToolResult,
)
MAX_MESSAGE_SIZE = 16 * 1024 * 1024 # 16MB
class ZapClient:
"""
ZAP Client for connecting to ZAP servers.
Example:
>>> async with ZapClient.connect("zap://localhost:9999") as client:
... tools = await client.list_tools()
... result = await client.call_tool("read_file", {"path": "README.md"})
"""
def __init__(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
self._reader = reader
self._writer = writer
self._server_info: ServerInfo | None = None
self._request_id = 0
@classmethod
async def connect(cls, url: str) -> ZapClient:
"""
Connect to a ZAP server.
Args:
url: Server URL (zap:// or zaps:// for TLS)
Returns:
Connected ZapClient instance
"""
parsed = urlparse(url)
use_tls = parsed.scheme == "zaps"
host = parsed.hostname or "localhost"
port = parsed.port or 9999
ssl_ctx = ssl.create_default_context() if use_tls else None
if ssl_ctx:
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
reader, writer = await asyncio.open_connection(host, port, ssl=ssl_ctx)
client = cls(reader, writer)
await client._handshake()
return client
async def _handshake(self) -> None:
"""Perform protocol handshake."""
client_info = ClientInfo(name="hanzo-zap", version="0.6.1")
await self._send(MessageType.INIT, client_info.__dict__)
msg_type, payload = await self._recv()
if msg_type != MessageType.INIT_ACK:
raise ConnectionError(f"Expected INIT_ACK, got {msg_type}")
self._server_info = ServerInfo(**payload)
@property
def server_info(self) -> ServerInfo | None:
"""Get server info from handshake."""
return self._server_info
async def list_tools(self) -> list[Tool]:
"""List available tools."""
await self._send(MessageType.LIST_TOOLS, {})
_, payload = await self._recv()
return [Tool(**t) for t in payload]
async def call_tool(self, name: str, args: dict[str, Any]) -> ToolResult:
"""Call a tool by name."""
self._request_id += 1
call = ToolCall(id=f"req-{self._request_id}", name=name, args=args)
await self._send(MessageType.CALL_TOOL, call.__dict__)
_, payload = await self._recv()
return ToolResult(**payload)
async def batch(
self, calls: list[dict[str, Any]]
) -> list[ToolResult]:
"""Call multiple tools in a batch."""
return [await self.call_tool(c["name"], c["args"]) for c in calls]
async def list_resources(self) -> list[Resource]:
"""List available resources."""
await self._send(MessageType.LIST_RESOURCES, {})
_, payload = await self._recv()
return [Resource(**r) for r in payload]
async def read_resource(self, uri: str) -> dict[str, Any]:
"""Read a resource by URI."""
await self._send(MessageType.READ_RESOURCE, {"uri": uri})
_, payload = await self._recv()
return payload
async def ping(self) -> None:
"""Send ping to check connection."""
await self._send(MessageType.PING, {})
msg_type, _ = await self._recv()
if msg_type != MessageType.PONG:
raise ConnectionError(f"Expected PONG, got {msg_type}")
async def close(self) -> None:
"""Close the connection."""
self._writer.close()
await self._writer.wait_closed()
async def __aenter__(self) -> ZapClient:
return self
async def __aexit__(self, *args: Any) -> None:
await self.close()
async def _send(self, msg_type: MessageType, payload: dict[str, Any]) -> None:
"""Send a message with ZAP wire format."""
payload_bytes = json.dumps(payload).encode("utf-8") if payload else b""
total_len = 1 + len(payload_bytes)
# Header: 4-byte LE length + 1-byte message type
header = struct.pack("<IB", total_len, msg_type)
self._writer.write(header + payload_bytes)
await self._writer.drain()
async def _recv(self) -> tuple[MessageType, Any]:
"""Receive and parse a message."""
# Read header
header = await self._reader.readexactly(5)
total_len, msg_type_byte = struct.unpack("<IB", header)
if total_len > MAX_MESSAGE_SIZE:
raise ValueError(f"Message too large: {total_len}")
# Read payload
payload_len = total_len - 1
if payload_len > 0:
payload_bytes = await self._reader.readexactly(payload_len)
payload = json.loads(payload_bytes.decode("utf-8"))
else:
payload = {}
msg_type = MessageType(msg_type_byte)
if msg_type == MessageType.ERROR:
raise RuntimeError(payload.get("message", "Server error"))
return msg_type, payload
View File
+187
View File
@@ -0,0 +1,187 @@
"""
ZAP Server implementation.
"""
from __future__ import annotations
import asyncio
import json
import struct
from collections.abc import Awaitable, Callable
from typing import Any
from .types import (
MessageType,
ServerInfo,
Tool,
ToolResult,
)
ToolHandler = Callable[[str, dict[str, Any]], Awaitable[Any] | Any]
class ZapServer:
"""
ZAP Server for hosting tools.
Example:
>>> server = ZapServer(name="my-tools", version="1.0.0")
>>> @server.tool("greet", "Greet someone")
... async def greet(name: str) -> str:
... return f"Hello, {name}!"
>>> await server.serve(9999)
"""
def __init__(self, name: str, version: str) -> None:
self._info = ServerInfo(
name=name,
version=version,
capabilities={"tools": True, "resources": False, "prompts": False},
)
self._tools: dict[str, tuple[Tool, ToolHandler]] = {}
self._server: asyncio.Server | None = None
def register_tool(
self,
name: str,
description: str,
input_schema: dict[str, Any],
handler: ToolHandler,
) -> None:
"""Register a tool."""
tool = Tool(name=name, description=description, input_schema=input_schema)
self._tools[name] = (tool, handler)
def tool(
self,
name: str,
description: str,
input_schema: dict[str, Any] | None = None,
) -> Callable[[ToolHandler], ToolHandler]:
"""Decorator to register a tool."""
def decorator(handler: ToolHandler) -> ToolHandler:
self.register_tool(
name=name,
description=description,
input_schema=input_schema or {},
handler=handler,
)
return handler
return decorator
async def serve(self, port: int, host: str = "0.0.0.0") -> None:
"""Start serving and block."""
self._server = await asyncio.start_server(
self._handle_connection, host, port
)
async with self._server:
await self._server.serve_forever()
async def start(self, port: int, host: str = "0.0.0.0") -> None:
"""Start serving in background."""
self._server = await asyncio.start_server(
self._handle_connection, host, port
)
await self._server.start_serving()
async def stop(self) -> None:
"""Stop the server."""
if self._server:
self._server.close()
await self._server.wait_closed()
async def _handle_connection(
self,
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
"""Handle a client connection."""
try:
while True:
# Read header
header = await reader.readexactly(5)
total_len, msg_type_byte = struct.unpack("<IB", header)
# Read payload
payload_len = total_len - 1
if payload_len > 0:
payload_bytes = await reader.readexactly(payload_len)
payload = json.loads(payload_bytes.decode("utf-8"))
else:
payload = {}
msg_type = MessageType(msg_type_byte)
await self._handle_message(writer, msg_type, payload)
except asyncio.IncompleteReadError:
pass # Connection closed
except Exception:
pass # Connection error
finally:
writer.close()
await writer.wait_closed()
async def _handle_message(
self,
writer: asyncio.StreamWriter,
msg_type: MessageType,
payload: dict[str, Any],
) -> None:
"""Handle a message from client."""
try:
if msg_type == MessageType.INIT:
await self._send(writer, MessageType.INIT_ACK, self._info.__dict__)
elif msg_type == MessageType.LIST_TOOLS:
tools = [t.__dict__ for t, _ in self._tools.values()]
await self._send(writer, MessageType.LIST_TOOLS_RESPONSE, tools)
elif msg_type == MessageType.CALL_TOOL:
result = await self._execute_tool(payload)
await self._send(writer, MessageType.CALL_TOOL_RESPONSE, result.__dict__)
elif msg_type == MessageType.PING:
await self._send(writer, MessageType.PONG, {})
else:
await self._send(
writer,
MessageType.ERROR,
{"message": f"Unknown message type: {msg_type}"},
)
except Exception as e:
await self._send(writer, MessageType.ERROR, {"message": str(e)})
async def _execute_tool(self, call: dict[str, Any]) -> ToolResult:
"""Execute a tool call."""
name = call.get("name", "")
args = call.get("args", {})
call_id = call.get("id", "")
if name not in self._tools:
return ToolResult(id=call_id, content=None, error=f"Unknown tool: {name}")
_, handler = self._tools[name]
try:
result = handler(name, args)
if asyncio.iscoroutine(result):
result = await result
return ToolResult(id=call_id, content=result)
except Exception as e:
return ToolResult(id=call_id, content=None, error=str(e))
async def _send(
self,
writer: asyncio.StreamWriter,
msg_type: MessageType,
payload: Any,
) -> None:
"""Send a message."""
payload_bytes = json.dumps(payload).encode("utf-8") if payload else b""
total_len = 1 + len(payload_bytes)
header = struct.pack("<IB", total_len, msg_type)
writer.write(header + payload_bytes)
await writer.drain()
+146
View File
@@ -0,0 +1,146 @@
"""
ZAP type definitions matching the Rust implementation.
"""
from dataclasses import dataclass, field
from enum import Enum, IntEnum
from typing import Any
class ApprovalPolicy(str, Enum):
"""Approval policy for tool execution (from hanzo-protocol)."""
UNLESS_TRUSTED = "unless-trusted" # Only auto-approve known-safe reads
ON_FAILURE = "on-failure" # Auto-approve, escalate on failure
ON_REQUEST = "on-request" # Model decides (default)
NEVER = "never" # Never ask
@dataclass
class SandboxPolicy:
"""Sandbox policy for tool execution (from hanzo-protocol)."""
mode: str # "danger-full-access", "read-only", "workspace-write"
writable_roots: list[str] = field(default_factory=list)
network_access: bool = False
allow_git_writes: bool = False
@classmethod
def danger_full_access(cls) -> "SandboxPolicy":
return cls(mode="danger-full-access")
@classmethod
def read_only(cls) -> "SandboxPolicy":
return cls(mode="read-only")
@classmethod
def workspace_write(
cls,
writable_roots: list[str] | None = None,
network_access: bool = False,
allow_git_writes: bool = False,
) -> "SandboxPolicy":
return cls(
mode="workspace-write",
writable_roots=writable_roots or [],
network_access=network_access,
allow_git_writes=allow_git_writes,
)
class MessageType(IntEnum):
"""Wire protocol message types."""
# Handshake
INIT = 0x01
INIT_ACK = 0x02
# Tools
LIST_TOOLS = 0x10
LIST_TOOLS_RESPONSE = 0x11
CALL_TOOL = 0x12
CALL_TOOL_RESPONSE = 0x13
# Resources
LIST_RESOURCES = 0x20
LIST_RESOURCES_RESPONSE = 0x21
READ_RESOURCE = 0x22
READ_RESOURCE_RESPONSE = 0x23
# Prompts
LIST_PROMPTS = 0x30
LIST_PROMPTS_RESPONSE = 0x31
GET_PROMPT = 0x32
GET_PROMPT_RESPONSE = 0x33
# Control
PING = 0xF0
PONG = 0xF1
ERROR = 0xFF
@dataclass
class Tool:
"""Tool definition."""
name: str
description: str
input_schema: dict[str, Any] = field(default_factory=dict)
@dataclass
class ToolCall:
"""Tool call request."""
id: str
name: str
args: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] | None = None
@dataclass
class ToolResult:
"""Tool execution result."""
id: str
content: Any = None
error: str | None = None
metadata: dict[str, Any] | None = None
@dataclass
class ServerInfo:
"""Server capabilities and info."""
name: str
version: str
capabilities: dict[str, bool] = field(
default_factory=lambda: {"tools": True, "resources": False, "prompts": False}
)
@dataclass
class ClientInfo:
"""Client info sent during handshake."""
name: str
version: str
@dataclass
class Resource:
"""Resource definition."""
uri: str
name: str
description: str | None = None
mime_type: str | None = None
@dataclass
class Prompt:
"""Prompt definition."""
name: str
description: str | None = None
arguments: list[dict[str, Any]] | None = None
+44
View File
@@ -0,0 +1,44 @@
[build-system]
requires = ["setuptools>=61.0.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "hanzo-zap"
version = "0.6.1"
description = "Zero-copy Agent Protocol (ZAP) SDK - 1000x faster than MCP"
readme = "README.md"
requires-python = ">=3.10"
license = { text = "MIT" }
authors = [{ name = "Hanzo AI", email = "[email protected]" }]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: Software Development :: Libraries :: Python Modules",
"Typing :: Typed",
]
keywords = ["agent", "ai", "hanzo", "mcp", "protocol", "zap", "zero-copy"]
dependencies = []
[project.urls]
Homepage = "https://github.com/hanzoai/python-sdk"
Repository = "https://github.com/hanzoai/python-sdk/tree/main/pkg/hanzo-zap"
Documentation = "https://hanzo.ai/docs/zap"
[project.optional-dependencies]
crypto = ["luxcrypto>=0.1.0"]
dev = [
"pytest>=7.0.0",
"pytest-asyncio>=0.26.0",
]
[tool.setuptools.packages.find]
where = ["."]
include = ["hanzo_zap*"]
[tool.setuptools.package-data]
hanzo_zap = ["py.typed"]
+1
View File
@@ -72,6 +72,7 @@ members = [
"pkg/hanzo-network",
"pkg/hanzo-repl",
"pkg/hanzo-s3",
"pkg/hanzo-zap",
"pkg/hanzoai",
]