pycapnp Documentation

Cap'n Proto supports a serialization mode that minimizes object copies by using segments. This is useful for high-throughput applications (e.g., sending data over ZeroMQ).

There are two ways to handle segments in Python:

1. to_segments(): Returns a list of copied, Python-owned bytes objects. Use this if you need the data to remain independent of the message builder's lifetime.
2. to_segment_views(): Returns read-only segment views that support the Python buffer protocol. These borrow memory from the message builder's arena. Warning: Do not mutate, reset, or reuse the builder while any segment view is still in use, as the views will become invalid.

Note: This feature is currently not supported in PyPy.

To reconstruct a message from segments, use from_segments(segments).

# Copying segments (safe, independent lifetime)
segments = alice.to_segments()
alice_reconstructed = addressbook_capnp.Person.from_segments(segments)

# Zero-copy segment views (high performance, requires builder to stay alive)
segment_views = alice.to_segment_views()
for segment in segment_views:
    transport.send(segment)  # segment supports buffer protocol

Once a client is created, you must bootstrap the server capability and cast it to the specific interface you want to use. Because capabilities are dynamic, you must use .cast_as(InterfaceName) to interpret them.

There are two ways to call methods:

1. Verbose syntax: Use .request_method_name() to set parameters individually and .send() to return a promise.
2. Short syntax: Pass a dictionary of arguments directly to the method name. This is more concise but can be difficult for deeply nested structures.

Pipelining: If a method returns a capability, you can access its methods immediately without awaiting the original promise. This allows chaining calls without extra network round-trips.

# 1. Bootstrap
calculator = client.bootstrap().cast_as(calculator_capnp.Calculator)

# 2. Verbose syntax
request = calculator.evaluate_request()
request.expression.literal = 123
eval_promise = request.send()

# 3. Short syntax
eval_promise = calculator.evaluate({"literal": 123})

# 4. Awaiting the result
result = await eval_promise()

# 5. Pipelining (accessing a capability returned by a method)
# If evaluate returns a capability named 'value':
read_promise = eval_promise.value.read()
read_result = await read_promise

To start a server, use capnp.AsyncIoStream.create_server. This method requires a callback function that is invoked whenever a new connection is made. The callback receives an AsyncIoStream instance.

Inside the callback, you should create a capnp.TwoPartyServer using the stream and your bootstrap implementation. It is recommended to await the .on_disconnect() method of the server to handle the connection lifetime properly.

import capnp
import asyncio

async def new_connection(stream):
    # bootstrap=YourImplementation() provides the server capabilities
    await capnp.TwoPartyServer(stream, bootstrap=CalculatorImpl()).on_disconnect()

async def main():
    host = 'localhost'
    port = '6000'
    server = await capnp.AsyncIoStream.create_server(new_connection, host, port)
    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(capnp.run(main()))

To implement a server, create a class that inherits from your_module_capnp.YourInterface.Server.

Rules for implementation:

• Method Names: Must match the interface exactly, or have _context appended to them.
• Standard Methods: If the name matches the interface, you receive arguments as keyword arguments (matching the spec) and a special _context parameter. It is recommended to include **kwargs to maintain compatibility if the interface evolves.
• Context Methods: If the name ends in _context, you only receive the context parameter. You must manually access context.params and set context.results.
• Return Values: If you return a promise, it is handled as a Cap'n Proto promise. Otherwise, the return statement values are mapped to the results struct in the order defined in the schema.

Example Schema:

interface TestInterface {
  foo @0 (i :UInt32, j :Bool) -> (x: Text, i:UInt32);
}

class TestInterface(capability_capnp.TestInterface.Server):
    # Standard method: arguments match spec, returns a tuple for results
    def foo(self, i, j, **kwargs):
        return str(j), i

    # Context method: manual param/result handling
    def defFunction_context(self, context):
        params = context.params
        context.results.func = FunctionImpl(params.paramCount, params.body)

You can write Cap'n Proto messages directly to file objects using write() or write_packed(). To read them back, use read() or read_packed().

Note that write() and read() are intended for files containing a single message. If your file contains multiple messages serialized sequentially, you must use the read_multiple() or read_multiple_packed() convenience functions to iterate over them.

# Single message serialization
addresses = addressbook_capnp.AddressBook.new_message()
...
with open('example.bin', 'wb') as f:
    addresses.write(f)

with open('example.bin', 'rb') as f:
    addresses = addressbook_capnp.AddressBook.read(f)

# Packed version
with open('example.bin', 'wb') as f:
    addresses.write_packed(f)

with open('example.bin', 'rb') as f:
    addresses = addressbook_capnp.AddressBook.read_packed(f)

# Multi-message files
with open('example.bin', 'wb') as f:
    addresses.write(f)
    addresses.write(f)
    addresses.write(f) # write 3 messages

with open('example.bin', 'rb') as f:
    for addresses in addressbook_capnp.AddressBook.read_multiple(f):
        print(addresses)

# Multi-message packed
for addresses in addressbook_capnp.AddressBook.read_multiple_packed(f):
    print(addresses)