### Install jsonrpcserver via pip

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/README.md

The installation command for the jsonrpcserver library using the Python package manager.

```shell
pip install jsonrpcserver
```

--------------------------------

### Define and Serve JSON-RPC Method in Python

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/jsonrpcserver.serve

This snippet shows how to define a simple 'ping' method using the @method decorator and start the server. Note that the serve function is deprecated and should be avoided in production environments.

```python
from jsonrpcserver import Result, Success, method, serve

@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")

if __name__ == "__main__":
    serve()
```

--------------------------------

### Built-in Development Server Usage

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Demonstrates how to use the built-in serve function to host a JSON-RPC server for development. It defines methods using the @method decorator and starts the server on a specified port.

```python
from jsonrpcserver import method, serve, Success, Result

@method
def ping() -> Result:
    return Success("pong")

@method
def add(a: int, b: int) -> Result:
    return Success(a + b)

if __name__ == "__main__":
    serve(port=5000)
```

--------------------------------

### Aiohttp Framework Integration

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Provides an example of using JSON-RPC with the aiohttp library. It leverages async_dispatch to handle asynchronous requests efficiently.

```python
from aiohttp import web
from jsonrpcserver import Result, Success, async_dispatch, method

@method
async def ping() -> Result:
    return Success("pong")

async def handle(request: web.Request) -> web.Response:
    return web.Response(
        text=await async_dispatch(await request.text()),
        content_type="application/json"
    )

app = web.Application()
app.router.add_post("/", handle)
```

--------------------------------

### JSON-RPC Method Definition

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Sanic

Example of defining a JSON-RPC method using the `@method` decorator.

```APIDOC
## JSON-RPC Method Definition

### Description
Defines a JSON-RPC method named `ping` that returns a success response with the value "pong".

### Method
(Internal - called by the dispatcher)

### Endpoint
(N/A)

### Parameters
#### Path Parameters
N/A

#### Query Parameters
N/A

#### Request Body
N/A

### Request Example
(N/A - this is a method definition)

### Response
#### Success Response
- **result** (string) - The success message, e.g., "pong".

#### Response Example
```json
{
  "jsonrpc": "2.0",
  "result": "pong",
  "id": 1
}
```
```

--------------------------------

### Creating Success Results with Success Function

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Demonstrates the use of the `Success` function to create successful JSON-RPC method results. It wraps return values in the Result Either type. Examples include returning data, lists, and handling void operations.

```python
from jsonrpcserver import method, dispatch, Success, Result

@method
def ping() -> Result:
    return Success("pong")

@method
def get_user(user_id: int) -> Result:
    user = {"id": user_id, "name": "Alice", "email": "alice@example.com"}
    return Success(user)

@method
def get_items() -> Result:
    items = ["apple", "banana", "cherry"]
    return Success(items)

@method
def void_operation() -> Result:
    return Success()

@method
def calculate_total(prices: list) -> Result:
    total = sum(prices)
    tax = total * 0.1
    return Success({"subtotal": total, "tax": tax, "total": total + tax})

response = dispatch('{"jsonrpc": "2.0", "method": "get_user", "params": [1], "id": 1}')

response = dispatch('{"jsonrpc": "2.0", "method": "calculate_total", "params": [[10.00, 25.50, 14.99]], "id": 2}')
```

--------------------------------

### Python: Create Custom JSON-RPC Error Results

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Demonstrates creating custom error results using the `Error` class for application-specific errors. It includes examples of division by zero, user not found, and invalid email format, showcasing custom error codes, messages, and optional data payloads. The `Success` class is used for valid responses.

```python
from jsonrpcserver import method, dispatch, Success, Error, Result

@method
def divide(a: float, b: float) -> Result:
    if b == 0:
        return Error(code=-32000, message="Division by zero")
    return Success(a / b)

@method
def get_user(user_id: int) -> Result:
    users = {1: "Alice", 2: "Bob"}
    if user_id not in users:
        return Error(
            code=-32001,
            message="User not found",
            data={"requested_id": user_id, "available_ids": list(users.keys())}
        )
    return Success({"id": user_id, "name": users[user_id]})

@method
def validate_email(email: str) -> Result:
    if "@" not in email:
        return Error(code=-32002, message="Invalid email format", data=email)
    return Success({"email": email, "valid": True})

# Usage
response = dispatch('{"jsonrpc": "2.0", "method": "divide", "params": [10, 0], "id": 1}')
# => '{"jsonrpc": "2.0", "error": {"code": -32000, "message": "Division by zero"}, "id": 1}'

response = dispatch('{"jsonrpc": "2.0", "method": "get_user", "params": [99], "id": 2}')
# => '{"jsonrpc": "2.0", "error": {"code": -32001, "message": "User not found", "data": {"requested_id": 99, "available_ids": [1, 2]}}, "id": 2}'
```

--------------------------------

### Python: Handle JSON-RPC Invalid Parameters

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Illustrates using the `InvalidParams` shortcut for standard JSON-RPC errors (code -32602). This is ideal for validating method arguments, as shown in examples for setting colors, creating users with specific constraints, and setting temperature with unit validation. It simplifies returning parameter-related errors.

```python
from jsonrpcserver import method, dispatch, Success, InvalidParams, Result

@method
def set_color(color: str) -> Result:
    valid_colors = ["red", "green", "blue", "yellow"]
    if color not in valid_colors:
        return InvalidParams(f"Invalid color '{color}'. Valid options: {valid_colors}")
    return Success({"color": color, "status": "set"})

@method
def create_user(name: str, age: int) -> Result:
    if not name or len(name) < 2:
        return InvalidParams("Name must be at least 2 characters")
    if age < 0 or age > 150:
        return InvalidParams(f"Age must be between 0 and 150, got {age}")
    return Success({"name": name, "age": age, "created": True})

@method
def set_temperature(value: float, unit: str) -> Result:
    if unit not in ["celsius", "fahrenheit", "kelvin"]:
        return InvalidParams()  # No data, just the standard error
    if unit == "kelvin" and value < 0:
        return InvalidParams("Kelvin cannot be negative")
    return Success({"temperature": value, "unit": unit})

# Usage
response = dispatch('{"jsonrpc": "2.0", "method": "set_color", "params": ["purple"], "id": 1}')
# => '{"jsonrpc": "2.0", "error": {"code": -32602, "message": "Invalid params", "data": "Invalid color \'purple\'. Valid options: [\'red\', \'green\', \'blue\', \'yellow\']"}, "id": 1}'

response = dispatch('{"jsonrpc": "2.0", "method": "create_user", "params": {"name": "A", "age": 25}, "id": 2}')
# => '{"jsonrpc": "2.0", "error": {"code": -32602, "message": "Invalid params", "data": "Name must be at least 2 characters"}, "id": 2}'
```

--------------------------------

### Python: Propagate JSON-RPC Errors via Exceptions

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Explains how to use the `JsonRpcError` exception to signal errors within method execution, particularly useful for aborting operations in nested calls. Examples include token validation, permission checks, processing payments, and handling invalid amounts or unsupported currencies, demonstrating how exceptions are converted into JSON-RPC error responses.

```python
from jsonrpcserver import method, dispatch, Success, JsonRpcError, Result

def validate_token(token: str) -> dict:
    """Helper function that validates authentication token."""
    if not token or len(token) < 10:
        raise JsonRpcError(code=-32001, message="Invalid token", data="Token too short")
    if token == "expired_token":
        raise JsonRpcError(code=-32002, message="Token expired")
    return {"user_id": 123, "role": "admin"}

def check_permissions(user: dict, resource: str) -> None:
    """Helper function that checks user permissions."""
    if user["role"] != "admin" and resource == "admin_panel":
        raise JsonRpcError(
            code=-32003,
            message="Permission denied",
            data={"required_role": "admin", "current_role": user["role"]}
        )

@method
def access_resource(token: str, resource: str) -> Result:
    # JsonRpcError propagates up and becomes an error response
    user = validate_token(token)
    check_permissions(user, resource)
    return Success({"resource": resource, "access": "granted", "user_id": user["user_id"]})

@method
def process_payment(amount: float, currency: str) -> Result:
    if amount <= 0:
        raise JsonRpcError(code=-32010, message="Invalid amount", data={"amount": amount})
    if currency not in ["USD", "EUR", "GBP"]:
        raise JsonRpcError(code=-32011, message="Unsupported currency", data=currency)
    return Success({"amount": amount, "currency": currency, "status": "processed"})

```

--------------------------------

### Python: Simplified Methods Initialization

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/CHANGELOG.md

The initialization API for the `Methods` object has been simplified. It now accepts functions directly as positional arguments or as keyword arguments with explicit names.

```python
from jsonrpcserver import Methods

# Initialize with positional arguments
methods_pos = Methods(func1, func2)

# Initialize with keyword arguments
methods_kw = Methods(name1=func1, name2=func2)
```

--------------------------------

### Implement JSON-RPC Server with Tornado

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Tornado

This snippet shows how to set up a basic Tornado application that routes POST requests to a JSON-RPC dispatcher. It includes a sample 'ping' method and the necessary request handler logic to process JSON-RPC payloads.

```python
from typing import Awaitable, Optional
from tornado import ioloop, web
from jsonrpcserver import Result, Success, async_dispatch, method

@method
async def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")

class MainHandler(web.RequestHandler):
    """Handle Tornado request"""
    async def post(self) -> None:
        """Post"""
        request = self.request.body.decode()
        response = await async_dispatch(request)
        if response:
            self.write(response)

    def data_received(self, chunk: bytes) -> Optional[Awaitable[None]]:
        pass

app = web.Application([(r"/", MainHandler)])

if __name__ == "__main__":
    app.listen(5000)
    ioloop.IOLoop.current().start()
```

--------------------------------

### Python: Configuration via .jsonrpcserverrc and Dispatch Arguments

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/CHANGELOG.md

The internal `config` module has been removed. Configuration is now handled externally via a `.jsonrpcserverrc` file or directly through arguments passed to the `dispatch` function.

```python
from jsonrpcserver import dispatch

# Configuration via dispatch arguments
response = dispatch(request, trim_log_values=True)

# Configuration via .jsonrpcserverrc file (content example):
# {
#   "trim_log_values": true
# }
```

--------------------------------

### Dispatching JSON-RPC Requests with Context

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Demonstrates how to dispatch JSON-RPC requests using `dispatch_to_response`, passing context to methods. It shows handling of parameters and disabling validation.

```python
from jsonrpcserver import dispatch_to_response, Success, Result

def get_user_info(context, user_id: int) -> Result:
    db = context["db"]
    return Success({"user_id": user_id, "session": context["session_id"]})

response = dispatch_to_response(
    '{"jsonrpc": "2.0", "method": "get_user_info", "params": [42], "id": 1}',
    methods={"get_user_info": get_user_info},
    context={"db": "database_connection", "session_id": "abc123"}
)

response = dispatch_to_response(
    '{"jsonrpc": "2.0", "method": "ping", "id": 1}',
    validator=lambda _: None  # Skip JSON-RPC schema validation
)
```

--------------------------------

### Asynchronous JSON-RPC Server with ZeroMQ (Python)

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Zeromq

Demonstrates an asynchronous JSON-RPC server using aiozmq and jsonrpcserver. This approach is suitable for high-concurrency applications. It uses asyncio and aiozmq to handle requests non-blockingly. Dependencies include 'aiozmq', 'zmq', and 'jsonrpcserver'.

```python
import asyncio

import aiozmq  # type: ignore
import zmq

from jsonrpcserver import Result, Success, async_dispatch, method


@method
async def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


async def main() -> None:
    """Handle AioZMQ request"""
    rep = await aiozmq.create_zmq_stream(zmq.REP, bind="tcp://*:5000")
    while True:
        request = (await rep.read())[0].decode()
        if response := (await async_dispatch(request)).encode():
            rep.write((response,))


if __name__ == "__main__":
    asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())
    asyncio.get_event_loop().run_until_complete(main())
```

--------------------------------

### Python: Basic HTTP Server Added

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/CHANGELOG.md

Version 3.3.0 introduced a basic HTTP server, simplifying the deployment and accessibility of the JSON-RPC service over the network.

```python
from jsonrpcserver import serve

if __name__ == "__main__":
    # Starts a basic HTTP server on port 5000
    serve(port=5000)
```

--------------------------------

### Define and dispatch JSON-RPC methods

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/README.md

Demonstrates how to register a method using the @method decorator and process a JSON-RPC request string using the dispatch function.

```python
from jsonrpcserver import dispatch, method, Success

@method
def ping():
    return Success("pong")

response = dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}')
# => '{"jsonrpc": "2.0", "result": "pong", "id": 1}'
```

--------------------------------

### Synchronous JSON-RPC Server with ZeroMQ (Python)

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Zeromq

Implements a basic synchronous JSON-RPC server using ZeroMQ. It binds to a TCP port, receives JSON-RPC requests, dispatches them using the jsonrpcserver library, and sends back the results. Dependencies include 'zmq' and 'jsonrpcserver'.

```python
import zmq

from jsonrpcserver import Result, Success, dispatch, method

socket = zmq.Context().socket(zmq.REP)


@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


if __name__ == "__main__":
    socket.bind("tcp://*:5000")
    while True:
        request = socket.recv().decode()
        socket.send_string(dispatch(request))
```

--------------------------------

### Python JSON-RPC Server using Werkzeug

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Werkzeug

This Python code sets up a basic JSON-RPC server using Werkzeug. It includes a 'ping' method for testing and a request handler that dispatches JSON-RPC calls. The server is configured to run on localhost:5000.

```python
from werkzeug.serving import run_simple
from werkzeug.wrappers import Request, Response

from jsonrpcserver import Result, Success, dispatch, method


@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


@Request.application
def application(request: Request) -> Response:
    """Handle Werkzeug request"""
    return Response(dispatch(request.data.decode()), 200, mimetype="application/json")


if __name__ == "__main__":
    run_simple("localhost", 5000, application)
```

--------------------------------

### FastAPI JSON-RPC Server Implementation

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Fast-API

This Python code demonstrates how to build a JSON-RPC server using FastAPI. It defines a simple 'ping' method and sets up an endpoint to handle JSON-RPC requests, returning responses via FastAPI's Response object. Dependencies include uvicorn, fastapi, and jsonrpcserver.

```python
import uvicorn
from fastapi import FastAPI, Request, Response

from jsonrpcserver import Result, Success, dispatch, method

app = FastAPI()


@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


@app.post("/")
async def index(request: Request) -> Response:
    """Handle FastAPI request"""
    return Response(dispatch(await request.body()))


if __name__ == "__main__":
    uvicorn.run(app, port=5000)
```

--------------------------------

### Create JSON-RPC Server with Aiohttp

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Aiohttp

This snippet shows how to define a JSON-RPC method and configure an aiohttp application to handle RPC requests. It uses the async_dispatch function to process incoming request bodies and return JSON-RPC compliant responses.

```python
from aiohttp import web
from jsonrpcserver import Result, Success, async_dispatch, method

@method
async def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")

async def handle(request: web.Request) -> web.Response:
    """Handle aiohttp request"""
    return web.Response(
        text=await async_dispatch(await request.text()), content_type="application/json"
    )

app = web.Application()
app.router.add_post("/", handle)

if __name__ == "__main__":
    web.run_app(app, port=5000)
```

--------------------------------

### WebSockets Integration for Real-time JSON-RPC

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Integrates the JSON RPC Server with the websockets library for real-time communication. It sets up a WebSocket server that handles incoming messages, dispatches them using async_dispatch, and sends back JSON-RPC responses. Requires websockets, asyncio, and jsonrpcserver. Supports asynchronous methods.

```python
import asyncio
from websockets.server import WebSocketServerProtocol, serve
from jsonrpcserver import Result, Success, async_dispatch, method

@method
async def ping() -> Result:
    return Success("pong")

@method
async def subscribe(channel: str) -> Result:
    return Success({"subscribed": channel, "status": "active"})

@method
async def broadcast(message: str) -> Result:
    return Success({"broadcast": message, "delivered": True})

async def handle(websocket: WebSocketServerProtocol, path: str) -> None:
    async for message in websocket:
        response = await async_dispatch(message)
        if response:  # Don't send empty responses for notifications
            await websocket.send(response)

async def main():
    async with serve(handle, "localhost", 5000):
        await asyncio.Future()  # Run forever

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

# Client example:
# import asyncio
# import websockets
# async def client():
#     async with websockets.connect("ws://localhost:5000") as ws:
#         await ws.send('{"jsonrpc": "2.0", "method": "ping", "id": 1}')
#         print(await ws.recv())
# asyncio.run(client())
```

--------------------------------

### Dispatch JSON-RPC requests

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Shows the usage of the dispatch function to process single, batch, and notification requests. It also demonstrates how to provide a custom methods dictionary.

```python
from jsonrpcserver import dispatch, method, Success, Result

@method
def ping() -> Result:
    return Success("pong")

@method
def sum_numbers(*args) -> Result:
    return Success(sum(args))

response = dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}')
response = dispatch('{"jsonrpc": "2.0", "method": "sum_numbers", "params": [1, 2, 3], "id": 2}')

# Batch request
response = dispatch('''[
    {"jsonrpc": "2.0", "method": "ping", "id": 1},
    {"jsonrpc": "2.0", "method": "sum_numbers", "params": [1, 2, 4], "id": 2}
]''')

# Manual methods dictionary
methods = {"multiply": lambda x, y: Success(x * y)}
response = dispatch('{"jsonrpc": "2.0", "method": "multiply", "params": [6, 7], "id": 1}', methods)
```

--------------------------------

### Sanic JSON-RPC Server Implementation (Python)

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Sanic

This Python code sets up a Sanic web server to handle JSON-RPC requests. It defines a simple 'ping' method and a route to dispatch incoming requests to the JSON-RPC dispatcher. Dependencies include Sanic and jsonrpcserver. It takes raw request body as input and returns JSON-RPC responses.

```python
from sanic import Sanic
from sanic.request import Request
from sanic.response import HTTPResponse, json

from jsonrpcserver import Result, Success, dispatch_to_serializable, method

app = Sanic("JSON-RPC app")


@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


@app.route("/", methods=["POST"]) 
async def test(request: Request) -> HTTPResponse:
    """Handle Sanic request"""
    return json(dispatch_to_serializable(request.body))


if __name__ == "__main__":
    app.run(port=5000)
```

--------------------------------

### Implement JSON-RPC over HTTPServer

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/HTTPServer

This snippet defines a custom BaseHTTPRequestHandler that reads incoming POST requests, dispatches them using the jsonrpcserver library, and returns the JSON-RPC response. It includes a sample 'ping' method and sets up a local server on port 5000.

```python
from http.server import BaseHTTPRequestHandler, HTTPServer
from jsonrpcserver import Result, Success, dispatch, method

@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")

class TestHttpServer(BaseHTTPRequestHandler):
    """HTTPServer request handler"""

    def do_POST(self) -> None:
        """POST handler"""
        request = self.rfile.read(int(self.headers["Content-Length"])).decode()
        response = dispatch(request)
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(response.encode())

if __name__ == "__main__":
    HTTPServer(("localhost", 5000), TestHttpServer).serve_forever()
```

--------------------------------

### Implement Aiozmq JSON-RPC Server

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Aiozmq

This snippet shows how to define a JSON-RPC method and run an asynchronous server using Aiozmq. It binds to a TCP port and continuously processes incoming requests using the async_dispatch function.

```python
import asyncio
import aiozmq
import zmq
from jsonrpcserver import Result, Success, async_dispatch, method

@method
async def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")

async def main() -> None:
    """Handle AioZMQ request"""
    rep = await aiozmq.create_zmq_stream(zmq.REP, bind="tcp://*:5000")
    while True:
        request = (await rep.read())[0].decode()
        if response := (await async_dispatch(request)).encode():
            rep.write((response,))

if __name__ == "__main__":
    asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())
    asyncio.get_event_loop().run_until_complete(main())
```

--------------------------------

### FastAPI Framework Integration

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Demonstrates integrating JSON-RPC with FastAPI. It utilizes the synchronous dispatch function within an asynchronous route handler.

```python
from fastapi import FastAPI, Request, Response
from jsonrpcserver import Result, Success, dispatch, method

app = FastAPI()

@method
def ping() -> Result:
    return Success("pong")

@app.post("/")
async def index(request: Request) -> Response:
    body = await request.body()
    return Response(dispatch(body), media_type="application/json")
```

--------------------------------

### Asynchronous JSON-RPC Request Dispatching

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Shows how to use `async_dispatch` for asynchronous request handling with async/await compatible frameworks. Methods can be defined as coroutines. Supports single and batch requests.

```python
import asyncio
from jsonrpcserver import async_dispatch, method, Success, Result

@method
async def ping() -> Result:
    return Success("pong")

@method
async def fetch_data(url: str) -> Result:
    await asyncio.sleep(0.1)
    return Success({"url": url, "status": "fetched"})

@method
async def parallel_sum(numbers: list) -> Result:
    await asyncio.sleep(0.01)
    return Success(sum(numbers))

async def main():
    response = await async_dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}')

    response = await async_dispatch('{"jsonrpc": "2.0", "method": "fetch_data", "params": ["https://api.example.com"], "id": 2}')

    response = await async_dispatch('''[
        {"jsonrpc": "2.0", "method": "ping", "id": 1},
        {"jsonrpc": "2.0", "method": "parallel_sum", "params": [[1, 2, 3, 4, 5]], "id": 2}
    ]''')

    return response

asyncio.run(main())
```

--------------------------------

### Python: Asyncio Support Added

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/CHANGELOG.md

Version 3.4.0 introduced support for `asyncio`, enabling asynchronous handling of JSON-RPC requests for Python 3.5 and later. This allows for non-blocking I/O operations.

```python
import asyncio
from jsonrpcserver import methods, serve

@methods.add
def async_task(delay):
    await asyncio.sleep(delay)
    return f"Task completed after {delay} seconds."

async def main():
    # Example of serving async requests
    await serve(port=5000)

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

--------------------------------

### Python Websockets JSON-RPC Server

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Websockets

This Python code sets up a basic websockets server that listens for incoming connections and dispatches JSON-RPC requests. It defines a simple 'ping' method and uses `async_dispatch` to handle requests. The server runs indefinitely on localhost:5000.

```Python
import asyncio

from websockets.server import WebSocketServerProtocol, serve

from jsonrpcserver import Result, Success, async_dispatch, method


@method
async def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


async def main(websocket: WebSocketServerProtocol, _: str) -> None:
    """Handle Websocket message"""
    if response := await async_dispatch(await websocket.recv()):
        await websocket.send(response)


start_server = serve(main, "localhost", 5000)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
```

--------------------------------

### Python: Method Dispatch with Methods Object

Source: https://github.com/explodinglabs/jsonrpcserver/blob/main/CHANGELOG.md

In version 4.0.0, the jsonrpcserver library mandates the use of the `Methods` object for dispatching. It no longer accepts dictionaries or lists directly. The `dispatch` method can now optionally use a global methods object if none is provided.

```python
from jsonrpcserver import Methods, dispatch

# New way to initialize Methods
methods = Methods(
    func1,
    func2,
    name1=func3,
    name2=func4
)

# Dispatching with a specific Methods object
response = dispatch(request, methods=methods)

# Dispatching using the global methods object (if not passed explicitly)
response = dispatch(request)
```

--------------------------------

### POST /dispatch

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Dispatches a JSON-RPC request string to the server, allowing for custom context injection and optional validation skipping.

```APIDOC
## POST /dispatch

### Description
Processes a JSON-RPC 2.0 request string. It supports injecting a context dictionary for dependency injection (e.g., database connections) and allows disabling schema validation for trusted inputs.

### Method
POST

### Endpoint
/dispatch

### Parameters
#### Request Body
- **json_request** (string) - Required - A valid JSON-RPC 2.0 request string.
- **methods** (dict) - Optional - A mapping of method names to callable functions.
- **context** (dict) - Optional - Data to be passed as the first argument to methods.
- **validator** (callable) - Optional - A function to override default schema validation.

### Request Example
{"jsonrpc": "2.0", "method": "get_user_info", "params": [42], "id": 1}

### Response
#### Success Response (200)
- **result** (object) - The execution result of the requested method.
- **id** (int) - The request ID matching the input.

#### Response Example
{"jsonrpc": "2.0", "result": {"user_id": 42, "session": "abc123"}, "id": 1}
```

--------------------------------

### POST /

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Aiohttp

The primary endpoint for handling JSON-RPC requests. It dispatches incoming JSON-RPC payloads to registered methods.

```APIDOC
## POST /

### Description
This endpoint receives JSON-RPC 2.0 requests, processes them using the `async_dispatch` function, and returns the corresponding JSON-RPC response.

### Method
POST

### Endpoint
/

### Request Body
- **jsonrpc** (string) - Required - Version of the JSON-RPC protocol (must be "2.0").
- **method** (string) - Required - The name of the method to be invoked (e.g., "ping").
- **params** (object) - Optional - Parameters for the method.
- **id** (string/number) - Optional - Request identifier.

### Request Example
{
  "jsonrpc": "2.0",
  "method": "ping",
  "id": 1
}

### Response
#### Success Response (200)
- **jsonrpc** (string) - Version of the JSON-RPC protocol.
- **result** (any) - The result of the method execution.
- **id** (string/number) - The request identifier matching the request.

#### Response Example
{
  "jsonrpc": "2.0",
  "result": "pong",
  "id": 1
}
```

--------------------------------

### Flask Framework Integration

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Shows how to integrate JSON-RPC with a Flask application. It uses the dispatch function to process raw request data within a POST route.

```python
from flask import Flask, Response, request
from jsonrpcserver import Result, Success, dispatch, method

app = Flask(__name__)

@method
def ping() -> Result:
    return Success("pong")

@app.route("/", methods=["POST"])
def index() -> Response:
    return Response(
        dispatch(request.get_data().decode()),
        content_type="application/json"
    )
```

--------------------------------

### POST /

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Tornado

The primary endpoint for receiving JSON-RPC requests. It processes the raw request body through the async_dispatch function and returns the JSON-RPC response.

```APIDOC
## POST /

### Description
Processes incoming JSON-RPC requests by dispatching them to registered methods and returning the resulting JSON-RPC response.

### Method
POST

### Endpoint
/

### Request Body
- **jsonrpc** (string) - Required - The version of the JSON-RPC protocol.
- **method** (string) - Required - The name of the method to be invoked.
- **params** (array/object) - Optional - The parameters to pass to the method.
- **id** (string/number) - Optional - An identifier established by the client.

### Request Example
{
  "jsonrpc": "2.0",
  "method": "ping",
  "id": 1
}

### Response
#### Success Response (200)
- **jsonrpc** (string) - The version of the JSON-RPC protocol.
- **result** (any) - The result of the method execution.
- **id** (string/number) - The identifier matching the request.

#### Response Example
{
  "jsonrpc": "2.0",
  "result": "pong",
  "id": 1
}
```

--------------------------------

### Dispatching to Serializable Dictionaries

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Illustrates the `dispatch_to_serializable` function, which returns Python dictionaries instead of JSON strings. This is useful for further processing before serialization. It supports single requests, batch requests, and notifications.

```python
from jsonrpcserver import dispatch_to_serializable, method, Success, Result

@method
def ping() -> Result:
    return Success("pong")

@method
def calculate(x: int, y: int) -> Result:
    return Success({"sum": x + y, "product": x * y})

response = dispatch_to_serializable('{"jsonrpc": "2.0", "method": "ping", "id": 1}')

response = dispatch_to_serializable('{"jsonrpc": "2.0", "method": "calculate", "params": [3, 4], "id": 2}')

response = dispatch_to_serializable('''[
    {"jsonrpc": "2.0", "method": "ping", "id": 1},
    {"jsonrpc": "2.0", "method": "calculate", "params": [5, 6], "id": 2}
]''')

response = dispatch_to_serializable('{"jsonrpc": "2.0", "method": "ping"}')
```

--------------------------------

### Python Flask JSON-RPC Server Implementation

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Flask

This Python code sets up a Flask web server to handle JSON-RPC requests. It defines a 'ping' method and uses the 'jsonrpcserver' library to dispatch incoming requests. The server listens for POST requests on the root URL and returns JSON-RPC responses.

```python
from flask import Flask, Response, request

from jsonrpcserver import Result, Success, dispatch, method

app = Flask(__name__)


@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


@app.route("/", methods=["POST"])
def index() -> Response:
    """Handle Flask request"""
    return Response(
        dispatch(request.get_data().decode()), content_type="application/json"
    )


if __name__ == "__main__":
    app.run()

```

--------------------------------

### Asyncio JSON-RPC Batch Request Handling (Python)

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Asyncio

This Python code snippet demonstrates how to process a batch of 100 JSON-RPC requests asynchronously using the 'jsonrpcserver' library and Python's 'asyncio'. It defines a simple asynchronous method 'sleep_' that pauses for 1 second, and then dispatches a JSON array containing 100 calls to this method.

```python
import asyncio
import json

from jsonrpcserver import Result, Success, async_dispatch, method


@method
async def sleep_() -> Result:
    """JSON-RPC method"""
    await asyncio.sleep(1)
    return Success()


async def handle(req: str) -> None:
    """Handle asyncio event"""
    print(await async_dispatch(req))


if __name__ == "__main__":
    request = json.dumps(
        [{"jsonrpc": "2.0", "method": "sleep_", "id": 1} for _ in range(100)]
    )
    asyncio.get_event_loop().run_until_complete(handle(request))
```

--------------------------------

### POST /async_dispatch

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Asynchronously dispatches JSON-RPC requests, supporting coroutine-based methods for high-concurrency environments.

```APIDOC
## POST /async_dispatch

### Description
An asynchronous version of the dispatch function, designed for use with async/await frameworks. It allows methods to be defined as `async def` for non-blocking I/O operations.

### Method
POST

### Endpoint
/async_dispatch

### Parameters
#### Request Body
- **request_data** (string) - Required - A JSON-RPC 2.0 request or batch request string.

### Request Example
{"jsonrpc": "2.0", "method": "fetch_data", "params": ["https://api.example.com"], "id": 2}

### Response
#### Success Response (200)
- **response** (string) - The serialized JSON-RPC response string.

#### Response Example
{"jsonrpc": "2.0", "result": {"url": "https://api.example.com", "status": "fetched"}, "id": 2}
```

--------------------------------

### POST / JSON-RPC Endpoint

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Sanic

This endpoint handles incoming JSON-RPC requests and dispatches them to the appropriate methods.

```APIDOC
## POST / JSON-RPC Endpoint

### Description
Handles incoming JSON-RPC requests by dispatching them to the appropriate methods using the `jsonrpcserver` library.

### Method
POST

### Endpoint
/

### Parameters
#### Request Body
- **(JSON)** - Required - The JSON-RPC request payload.

### Request Example
```json
{
  "jsonrpc": "2.0",
  "method": "ping",
  "params": [],
  "id": 1
}
```

### Response
#### Success Response (200)
- **(JSON)** - The JSON-RPC response payload.

#### Response Example
```json
{
  "jsonrpc": "2.0",
  "result": "pong",
  "id": 1
}
```
```

--------------------------------

### Register methods with @method decorator

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Demonstrates how to register Python functions as JSON-RPC methods using the @method decorator. Supports custom names, positional arguments, and keyword arguments.

```python
from jsonrpcserver import method, dispatch, Success, Result

@method
def ping() -> Result:
    return Success("pong")

@method(name="math.add")
def add_numbers(a: int, b: int) -> Result:
    return Success(a + b)

@method
def subtract(minuend: int, subtrahend: int) -> Result:
    return Success(minuend - subtrahend)

@method
def greet(**kwargs) -> Result:
    name = kwargs.get("name", "World")
    return Success(f"Hello, {name}!")

response = dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}')
response = dispatch('{"jsonrpc": "2.0", "method": "math.add", "params": [5, 3], "id": 2}')
response = dispatch('{"jsonrpc": "2.0", "method": "subtract", "params": {"minuend": 42, "subtrahend": 23}, "id": 3}')
```

--------------------------------

### POST /jsonrpc (Batch Request)

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Asyncio

Processes a batch of JSON-RPC requests asynchronously. This endpoint allows multiple method calls to be executed concurrently, improving performance for I/O-bound tasks.

```APIDOC
## POST /jsonrpc

### Description
Processes a batch of JSON-RPC 2.0 requests asynchronously. The server executes the specified methods concurrently and returns an array of results.

### Method
POST

### Endpoint
/jsonrpc

### Parameters
#### Request Body
- **batch** (array) - Required - An array of JSON-RPC request objects, each containing 'jsonrpc', 'method', and 'id'.

### Request Example
[
  {"jsonrpc": "2.0", "method": "sleep_", "id": 1},
  {"jsonrpc": "2.0", "method": "sleep_", "id": 2}
]

### Response
#### Success Response (200)
- **result** (array) - An array of JSON-RPC response objects corresponding to the input batch.

#### Response Example
[
  {"jsonrpc": "2.0", "result": null, "id": 1},
  {"jsonrpc": "2.0", "result": null, "id": 2}
]
```

--------------------------------

### Django Integration for JSON-RPC Requests

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Integrates the JSON RPC Server with Django views using the dispatch function. It handles incoming HTTP requests, decodes the JSON-RPC payload, and returns a JSON-RPC response. Requires Django and jsonrpcserver. Handles CSRF exemption for RPC endpoints.

```python
from django.http import HttpRequest, HttpResponse
from django.views.decorators.csrf import csrf_exempt
from jsonrpcserver import Result, Success, Error, dispatch, method

@method
def ping() -> Result:
    return Success("pong")

@method
def create_item(name: str, price: float) -> Result:
    if price < 0:
        return Error(-32000, "Price cannot be negative")
    return Success({"name": name, "price": price, "created": True})

@csrf_exempt
def jsonrpc(request: HttpRequest) -> HttpResponse:
    return HttpResponse(
        dispatch(request.body.decode()),
        content_type="application/json"
    )

# In urls.py:
# from django.urls import path
# from .views import jsonrpc
# urlpatterns = [
#     path('rpc/', jsonrpc),
# ]
```

--------------------------------

### Python Flask-SocketIO JSON-RPC Server

Source: https://github.com/explodinglabs/jsonrpcserver/wiki/Socketio

This Python code implements a JSON-RPC server using Flask and Flask-SocketIO. It defines a 'ping' method and handles incoming messages by dispatching them. The server listens on port 5000.

```python
from flask import Flask, Request
from flask_socketio import SocketIO, send  # type: ignore

from jsonrpcserver import Result, Success, dispatch, method

app = Flask(__name__)
socketio = SocketIO(app)


@method
def ping() -> Result:
    """JSON-RPC method"""
    return Success("pong")


@socketio.on("message")  # type: ignore
def handle_message(request: Request) -> None:
    """Handle SocketIO request"""
    if response := dispatch(request):
        send(response, json=True)


if __name__ == "__main__":
    socketio.run(app, port=5000)
```

--------------------------------

### JsonRpcError Exception

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Explains how to use the JsonRpcError exception to return errors from deep within method calls, allowing for cleaner error propagation.

```APIDOC
## JsonRpcError Exception

The `JsonRpcError` exception can be raised inside a method as an alternative way to return an error response. This is useful when you want to abort execution deep in nested function calls.

### Method Signature
`JsonRpcError(code: int, message: str, data: any = None)`

### Parameters
- **code** (int) - Required - The error code.
- **message** (str) - Required - A short description of the error.
- **data** (any) - Optional - Additional data related to the error.

### Example Usage
```python
from jsonrpcserver import method, dispatch, Success, JsonRpcError, Result

def validate_token(token: str) -> dict:
    """Helper function that validates authentication token."""
    if not token or len(token) < 10:
        raise JsonRpcError(code=-32001, message="Invalid token", data="Token too short")
    if token == "expired_token":
        raise JsonRpcError(code=-32002, message="Token expired")
    return {"user_id": 123, "role": "admin"}

def check_permissions(user: dict, resource: str) -> None:
    """Helper function that checks user permissions."""
    if user["role"] != "admin" and resource == "admin_panel":
        raise JsonRpcError(
            code=-32003,
            message="Permission denied",
            data={"required_role": "admin", "current_role": user["role"]}
        )

@method
def access_resource(token: str, resource: str) -> Result:
    # JsonRpcError propagates up and becomes an error response
    user = validate_token(token)
    check_permissions(user, resource)
    return Success({"resource": resource, "access": "granted", "user_id": user["user_id"]})

@method
def process_payment(amount: float, currency: str) -> Result:
    if amount <= 0:
        raise JsonRpcError(code=-32010, message="Invalid amount", data={"amount": amount})
    if currency not in ["USD", "EUR", "GBP"]:
        raise JsonRpcError(code=-32011, message="Unsupported currency", data=currency)
    return Success({"amount": amount, "currency": currency, "status": "processed"})

# Example of raising JsonRpcError
try:
    response = dispatch('{"jsonrpc": "2.0", "method": "access_resource", "params": {"token": "short", "resource": "admin_panel"}, "id": 1}')
    # Expected Response: '{"jsonrpc": "2.0", "error": {"code": -32001, "message": "Invalid token", "data": "Token too short"}, "id": 1}'
except Exception as e:
    print(e)

try:
    response = dispatch('{"jsonrpc": "2.0", "method": "process_payment", "params": [100, "JPY"], "id": 2}')
    # Expected Response: '{"jsonrpc": "2.0", "error": {"code": -32011, "message": "Unsupported currency", "data": "JPY"}, "id": 2}'
except Exception as e:
    print(e)
```
```

--------------------------------

### POST /dispatch_to_serializable

Source: https://context7.com/explodinglabs/jsonrpcserver/llms.txt

Dispatches a JSON-RPC request and returns a Python dictionary or list of dictionaries instead of a serialized JSON string.

```APIDOC
## POST /dispatch_to_serializable

### Description
Processes a JSON-RPC request and returns the raw Python dictionary representation. This is useful for middleware or scenarios where further processing is required before final serialization.

### Method
POST

### Endpoint
/dispatch_to_serializable

### Parameters
#### Request Body
- **request_data** (string) - Required - A JSON-RPC 2.0 request or batch request string.

### Request Example
{"jsonrpc": "2.0", "method": "ping", "id": 1}

### Response
#### Success Response (200)
- **response** (dict|list|None) - The result as a Python object.

#### Response Example
{"jsonrpc": "2.0", "result": "pong", "id": 1}
```