Skip to main content

Nuncio Communication Protocol Core Python SDK

Project description

Nuncio Python SDK (nuncio)

License: Apache-2.0 Python Versions

Python SDK for the Nuncio Communication Protocol—the open-source, cooperative peer-to-peer agent communication standard.


🌟 Overview

Nuncio makes agent‑to‑agent and tool-using communication simple, bi-directional, and robust. It gives you a clean, versioned binary envelope, a predictable handshake, and native patterns for routing, negotiation, text streaming, and errors—all over WebSockets using Google Protocol Buffers (v3).

This SDK implements the Nuncio specification, making it easy to:

  • Build Nuncio Clients that connect to any peer or router.
  • Create Nuncio Server Routers that coordinate and route messages between agents.
  • Utilize Standardized Envelopes for RPC execution, streaming chunks, capability negotiation, and peer discovery.
  • Suspend Workflows cleanly using native human-in-the-loop approval intercepts.

📦 Installation

Install the official package via pip:

pip install nuncio

Or using uv:

uv pip install nuncio

🚀 Quickstart

Get a P2P agent connection running in less than a minute.

1. Start the Nuncio Router/Server

Create a server that routes envelopes to agents based on their specific roles (e.g. RESEARCH_AGENT):

from nuncio.server import NuncioServer

# Initialize the router server
server = NuncioServer(host="127.0.0.1", port=8080)
print("Starting Nuncio Router on port 8080...")
server.run()

2. Connect Client A (The Tool-Providing Responder)

This client registers as RESEARCH_AGENT and responds to requests:

import asyncio
from nuncio.client import NuncioClient, NuncioBuilders
from nuncio.nuncio_pb2 import MessageType

async def main() -> None:
    # Initialize connection registration
    client = NuncioClient(
        uri="ws://localhost:8080/nuncio",
        role="RESEARCH_AGENT",
        agent_id="research-alpha",
        token="demo-token",
        capabilities=["web_search", "code_eval"],
    )
    await client.connect()
    print("Research Agent online & waiting for tasks...")

    while True:
        incoming = await client.recv_envelope()
        
        # Respond to requests
        if incoming.type == MessageType.REQUEST:
            print(f"Received Request: {incoming.request.intent} (ID: {incoming.request.request_id})")
            
            response = NuncioBuilders.response(
                intent=incoming.request.intent,
                request_id=incoming.request.request_id,
                content="Search results: Nuncio is an open agent protocol.",
                confidence=0.98,
            )
            
            # Send back the correlated response envelope
            env = client.build_envelope(MessageType.RESPONSE)
            env.response.CopyFrom(response)
            await client.send_envelope(env)

asyncio.run(main())

3. Connect Client B (The Coordinating Requester)

This client connects as COORDINATOR_AGENT and requests tools/delegations from any available research agent on the network:

import asyncio
from nuncio.client import NuncioClient, NuncioBuilders
from nuncio.nuncio_pb2 import MessageType

async def main() -> None:
    client = NuncioClient(
        uri="ws://localhost:8080/nuncio",
        role="COORDINATOR_AGENT",
        agent_id="coordinator-main",
        token="demo-token",
        capabilities=[],
    )
    await client.connect()
    print("Coordinator Agent online.")

    # Target any agent matching the logical role 'RESEARCH_AGENT'
    request = NuncioBuilders.request(intent="search_agent_protocol")
    env = client.build_envelope(MessageType.REQUEST)
    env.request.CopyFrom(request)
    env.target_role = "RESEARCH_AGENT"  # Symmetric routing target
    
    await client.send_envelope(env)
    print("Task dispatched...")

    # Wait for the bi-directional response
    incoming = await client.recv_envelope()
    if incoming.type == MessageType.RESPONSE:
        print(f"Response Received: {incoming.response.content}")
    elif incoming.type == MessageType.ERROR:
        print(f"Error Received: {incoming.error.message}")

asyncio.run(main())

⚖️ License

Nuncio is open-source software licensed under the Apache License 2.0. See the LICENSE file for details.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

nuncio-0.1.3.tar.gz (23.6 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

nuncio-0.1.3-py3-none-any.whl (23.4 kB view details)

Uploaded Python 3

File details

Details for the file nuncio-0.1.3.tar.gz.

File metadata

  • Download URL: nuncio-0.1.3.tar.gz
  • Upload date:
  • Size: 23.6 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.11

File hashes

Hashes for nuncio-0.1.3.tar.gz
Algorithm Hash digest
SHA256 4da1fb2e2234c4c94b4ea182745e570c59ee81447694173551d95d49d5d0e328
MD5 3a87a41cd5919d76befe8de6ed8e4bc0
BLAKE2b-256 002057047bee2bca398aa1ae960e1a222e4c7362f07cc52ee3d8d3e25a1999f0

See more details on using hashes here.

File details

Details for the file nuncio-0.1.3-py3-none-any.whl.

File metadata

  • Download URL: nuncio-0.1.3-py3-none-any.whl
  • Upload date:
  • Size: 23.4 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.11

File hashes

Hashes for nuncio-0.1.3-py3-none-any.whl
Algorithm Hash digest
SHA256 88a670f3d608299373d4004ec361f84a604250720b09baaf2405de87da1165a0
MD5 1a61ebd861f256cbf34d479805ab6ba9
BLAKE2b-256 5fc03edfe10a282604bdd3bebfd3113e0a9f413987e32c70c3dc0ff74748395d

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page