Install & Compatibility
Where this runs
tested against v1.2.0 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.95 runs
installs and imports cleanly · install 0.0s · import 0.218s · 18MB
glibcpy 3.10–3.95 runs
installs and imports cleanly · install 1.5s · import 0.196s · 19MB
16MB installed
● package 16MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
AsyncJSONRPCResponseManager
✓ from ajsonrpc import AsyncJSONRPCResponseManager
✗ from ajsonrpc.manager import JSONRPCResponseManager
Dispatcher
✓ from ajsonrpc import Dispatcher
core
✓ from ajsonrpc import core
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/
Upgrade
Version history
1.2.0latest on PyPI · released Jul 21, 2021
Audit
Dependencies
No dependency data recorded yet.