ajsonrpc

1.2.0 · active · verified Sat Apr 11

ajsonrpc is an active, lightweight Python library (version 1.2.0) that provides an asynchronous implementation of the JSON-RPC 2.0 protocol and server capabilities, powered by asyncio. It functions as a successor to the `json-rpc` library, offering largely compatible code. It focuses on being vanilla Python with no external dependencies for its core functionality and has a release cadence driven by feature additions like new backend support.

Warnings

breaking Major interface changes occurred between pre-1.0.0 versions (0.x.x) and the stable 1.0.0 release. The 0.1.0 release was explicitly marked with an 'unstable interface,' and 1.0.0a was a 'Freezing interface' release. Users upgrading from older 0.x.x versions should review their code, especially around protocol handling and manager initialization, as direct compatibility is not guaranteed.
Fix: Refer to the GitHub repository's `ajsonrpc` and `json-rpc` documentation and examples for updated patterns, particularly regarding the `Dispatcher` and `JSONRPCResponseManager` usage.
gotcha By default, `JSONRPCResponseManager` returns a generic `ServerError` without detailed exception information for security reasons. If your application relies on verbose error reporting (e.g., for debugging in non-production environments), you must explicitly configure the manager to include exception details.
Fix: When initializing `JSONRPCResponseManager`, set `verbose_exceptions=True` (e.g., `manager = JSONRPCResponseManager(dispatcher, verbose_exceptions=True)`).
gotcha While `ajsonrpc` is designed for asynchronous operations, its core components like `JSONRPCResponseManager` can be misused in synchronous contexts. All methods intended for handling incoming RPC payloads (e.g., `get_payload_for_payload`) are `async` functions and must be `await`ed. Attempting to call them synchronously will lead to runtime errors or unexpected behavior.
Fix: Always use `await` when calling asynchronous methods of `ajsonrpc` components, ensuring your server loop and request handling are fully `async` compatible.

Install

pip install ajsonrpc Install latest version

Imports

JSONRPCResponseManager

from ajsonrpc.manager import JSONRPCResponseManager

Dispatcher

from ajsonrpc.dispatcher import Dispatcher

JSONRPCServerProtocol

from ajsonrpc.protocol import JSONRPCServerProtocol

AiohttpJSONRPCServer
```
from ajsonrpc.server.aiohttp import AiohttpJSONRPCServer
```
For integrating with aiohttp framework
QuartJSONRPCServer
```
from ajsonrpc.server.quart import QuartJSONRPCServer
```
For integrating with Quart framework (added in v1.2.0)
TornadoJSONRPCServer
```
from ajsonrpc.server.tornado import TornadoJSONRPCServer
```
For integrating with Tornado framework

Quickstart

This quickstart demonstrates how to set up a basic `ajsonrpc` server using `asyncio.start_server`. It defines a `Dispatcher` to register RPC methods (both synchronous and asynchronous) and uses a `JSONRPCResponseManager` to handle incoming requests and generate responses. A custom `JSONRPCServerProtocol` bridges the `asyncio` transport with the `ajsonrpc` manager.

import asyncio
from ajsonrpc.manager import JSONRPCResponseManager
from ajsonrpc.dispatcher import Dispatcher
from ajsonrpc.protocol import JSONRPCServerProtocol

class MyJSONRPCServerProtocol(JSONRPCServerProtocol):
    def __init__(self, manager):
        super().__init__()
        self._manager = manager

    async def handle_request(self, payload):
        # The manager handles parsing, dispatching to methods, and error handling
        return await self._manager.get_payload_for_payload(payload)

def get_dispatcher():
    dispatcher = Dispatcher()

    @dispatcher.add_method
    def add(a, b):
        """Adds two numbers."""
        return a + b

    @dispatcher.add_method
    async def say_hello(name="World"):
        """Asynchronously greets a given name."""
        await asyncio.sleep(0.1) # Simulate async work
        return f"Hello, {name}!"

    return dispatcher

async def main():
    dispatcher = get_dispatcher()
    manager = JSONRPCResponseManager(dispatcher)

    # Create a server factory that passes the manager instance
    server_factory = lambda: MyJSONRPCServerProtocol(manager)
    
    # Start an asyncio TCP server
    server = await asyncio.start_server(server_factory, '127.0.0.1', 8080)

    addr = server.sockets[0].getsockname()
    print(f"Serving JSON-RPC on {addr}. Press Ctrl+C to stop.")

    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nServer stopped.")

# Example usage (run this in another terminal while the server is running):
# curl -X POST -H "Content-Type: application/json" \
#      -d '{"jsonrpc": "2.0", "method": "add", "params": [1, 2], "id": 1}' \
#      http://127.0.0.1:8080/
#
# curl -X POST -H "Content-Type: application/json" \
#      -d '{"jsonrpc": "2.0", "method": "say_hello", "params": {"name": "Async User"}, "id": 2}' \
#      http://127.0.0.1:8080/
#
# curl -X POST -H "Content-Type: application/json" \
#      -d '{"jsonrpc": "2.0", "method": "non_existent_method", "params": [], "id": 3}' \
#      http://127.0.0.1:8080/

view raw JSON →