pip install zap-protocol

# Or with uv
uv add zap-protocol

import asyncio
import json
import uuid
from zap_protocol import Zap, ClientInfo, ToolCall

async def main():
    # Connect to ZAP server
    client = await Zap.connect("zap://localhost:9000")

    # Initialize
    server = await client.init(ClientInfo(
        name="my-agent",
        version="1.0.0"
    ))
    print(f"Connected to: {server.name} v{server.version}")

    # List tools
    tools = await client.list_tools()
    for tool in tools:
        print(f"  {tool.name} - {tool.description}")

    # Call a tool
    result = await client.call_tool(ToolCall(
        id=str(uuid.uuid4()),
        name="read_file",
        args=json.dumps({"path": "/etc/hosts"}).encode()
    ))

    if result.error:
        print(f"Error: {result.error}")
    else:
        print(f"Result: {result.content.decode()}")

    await client.close()

asyncio.run(main())

# Simple connection
client = await Zap.connect("zap://localhost:9000")

# With options
client = await Zap.connect(
    "zap://localhost:9000",
    timeout=30.0,
    identity=identity,
)

# Context manager
async with Zap.connect("zap://localhost:9000") as client:
    tools = await client.list_tools()

# List tools
tools = await client.list_tools()

# Call tool
result = await client.call_tool(ToolCall(
    id="call-1",
    name="my_tool",
    args=b'{"param": "value"}'
))

# With typed result
content = json.loads(result.content)

# List resources
resources = await client.list_resources()

# Read resource
content = await client.read_resource("file:///etc/hosts")
print(content.text)  # or content.blob for binary

# Subscribe to updates
stream = await client.subscribe("file:///var/log/app.log")
async for update in stream:
    print(f"Update: {update.content}")

# List prompts
prompts = await client.list_prompts()

# Get prompt messages
messages = await client.get_prompt(
    "code_review",
    {"topic": "python"}
)
for msg in messages:
    print(f"{msg.role}: {msg.content}")

from zap_protocol import Gateway, ServerConfig, Transport, Auth
import os

# Create gateway
gateway = Gateway()

# Add MCP server
fs_id = await gateway.add_server(
    "filesystem",
    "stdio://npx -y @modelcontextprotocol/server-filesystem /tmp",
    ServerConfig(transport=Transport.STDIO)
)

# Add with authentication
gh_id = await gateway.add_server(
    "github",
    "stdio://npx -y @modelcontextprotocol/server-github",
    ServerConfig(
        transport=Transport.STDIO,
        auth=Auth(bearer=os.environ["GITHUB_TOKEN"]),
        timeout=60000
    )
)

# List all tools
tools = await gateway.list_tools()

# Check status
status = await gateway.server_status(fs_id)
print(f"Status: {status}")

from zap_protocol import (
    Zap,
    Tool,
    ToolCall,
    ToolResult,
    Resource,
    ResourceContent,
    ServerInfo,
    ClientInfo,
)

async def process_tools(client: Zap) -> list[str]:
    tools: list[Tool] = await client.list_tools()
    return [t.name for t in tools]

from zap_protocol import ZapError, DisconnectedError, TimeoutError

try:
    result = await client.call_tool(call)
except DisconnectedError:
    print("Connection lost, reconnecting...")
    await client.reconnect()
except TimeoutError:
    print("Request timed out")
except ZapError as e:
    print(f"ZAP error: {e}")

if result.error:
    print(f"Tool error: {result.error}")

from zap_protocol.server import Server, ZapHandler
from zap_protocol import (
    ServerInfo,
    ClientInfo,
    Tool,
    ToolCall,
    ToolResult,
    Capabilities,
)
import json

class MyHandler(ZapHandler):
    async def init(self, client: ClientInfo) -> ServerInfo:
        return ServerInfo(
            name="my-server",
            version="1.0.0",
            capabilities=Capabilities(
                tools=True,
                resources=True,
                prompts=False,
                logging=True
            )
        )

    async def list_tools(self) -> list[Tool]:
        return [
            Tool(
                name="hello",
                description="Says hello",
                schema=json.dumps({
                    "type": "object",
                    "properties": {"name": {"type": "string"}}
                }).encode()
            )
        ]

    async def call_tool(self, call: ToolCall) -> ToolResult:
        if call.name == "hello":
            args = json.loads(call.args)
            name = args.get("name", "World")
            return ToolResult(
                id=call.id,
                content=f"Hello, {name}!".encode()
            )
        return ToolResult(
            id=call.id,
            error="Unknown tool"
        )

async def main():
    server = Server(MyHandler())
    await server.serve("0.0.0.0", 9000)

asyncio.run(main())

import asyncio

async def call_tools_concurrently(client: Zap, calls: list[ToolCall]):
    results = await asyncio.gather(*[
        client.call_tool(call) for call in calls
    ])
    return results

async def monitor_resource(client: Zap, uri: str):
    stream = await client.subscribe(uri)
    async for update in stream:
        print(f"Update: {update.content}")

# Run in background
task = asyncio.create_task(monitor_resource(client, "file:///log"))

async def with_timeout():
    try:
        async with asyncio.timeout(10.0):
            result = await client.call_tool(call)
    except asyncio.TimeoutError:
        print("Call timed out")

import pytest
from unittest.mock import AsyncMock
from zap_protocol import Zap, Tool

@pytest.fixture
def mock_client():
    client = AsyncMock(spec=Zap)
    client.list_tools.return_value = [
        Tool(name="test", description="Test tool", schema=b"{}")
    ]
    return client

@pytest.mark.asyncio
async def test_list_tools(mock_client):
    tools = await mock_client.list_tools()
    assert len(tools) == 1
    assert tools[0].name == "test"