fgp-daemon-py

Python SDK for building Fast Gateway Protocol (FGP) daemons.

FGP daemons use UNIX sockets with NDJSON framing to achieve 10-30ms response times, compared to 200-500ms for stdio-based protocols like MCP.

Installation

pip install fgp-daemon

Or install from source:

git clone https://github.com/fast-gateway-protocol/daemon-py
cd fgp-daemon-py
pip install -e .

Quick Start

from fgp_daemon import FgpServer, FgpService
from typing import Any

class MyService(FgpService):
    def name(self) -> str:
        return "my-service"

    def version(self) -> str:
        return "1.0.0"

    def dispatch(self, method: str, params: dict[str, Any]) -> Any:
        if method == "my-service.echo":
            return {"echo": params}
        raise ValueError(f"Unknown method: {method}")

server = FgpServer(MyService(), "~/.fgp/services/my-service/daemon.sock")
server.serve()

Features

• Fast: 10-30ms response times via persistent UNIX sockets
• Simple: NDJSON protocol (one JSON object per line)
• Built-in methods: health, stop, methods handled automatically
• Lifecycle hooks: on_start() and on_stop() for setup/cleanup
• Health checks: Custom health status reporting
• Thin client: Call any FGP daemon from Python

Protocol

Request:

{"id":"uuid","v":1,"method":"service.action","params":{}}

Response:

{"id":"uuid","ok":true,"result":{},"meta":{"server_ms":12,"protocol_v":1}}

See FGP-PROTOCOL.md for the full specification.

Examples

Run the echo daemon example:

python examples/echo_daemon.py

Test with the fgp CLI:

fgp call echo.echo -p '{"message": "hello"}'
fgp health echo
fgp methods echo

Or with netcat:

echo '{"id":"1","v":1,"method":"health","params":{}}' | nc -U ~/.fgp/services/echo/daemon.sock

Client Usage

from fgp_daemon import FgpClient

client = FgpClient("~/.fgp/services/gmail/daemon.sock")

# Call a method
response = client.call("gmail.list", {"limit": 10})
if response.ok:
    print(response.result)
else:
    print(f"Error: {response.error.message}")

# Built-in convenience methods
health = client.health()
methods = client.methods()
client.stop()

Service Implementation

Method List

Provide introspection for your service's methods:

from fgp_daemon.service import MethodInfo, ParamInfo

def method_list(self) -> list[MethodInfo]:
    return [
        MethodInfo(
            name="my-service.list",
            description="List items",
            params=[
                ParamInfo(name="limit", param_type="int", required=False, default=10),
            ],
        ),
    ]

Health Checks

Report detailed health status:

from fgp_daemon.service import HealthStatus

def health_check(self) -> dict[str, HealthStatus]:
    return {
        "database": HealthStatus(ok=True, message="Connected"),
        "cache": HealthStatus(ok=redis.ping(), message="Redis available"),
    }

Lifecycle Hooks

def on_start(self) -> None:
    # Initialize connections, load config, warm caches
    self.db = connect_to_database()

def on_stop(self) -> None:
    # Clean up resources
    self.db.close()

Related Projects

• protocol - FGP Protocol Specification
• daemon - Rust SDK (faster for performance-critical daemons)
• cli - CLI for managing FGP daemons

License

MIT