Metadata-Version: 2.5
Name: pp-rpc-server
Version: 0.1.0
Summary: Experimental RPC server for remote pi sessions over framed CBOR bytes (Python port of @earendil-works/pi-server)
Project-URL: Homepage, https://github.com/HSPK/pp_rpc_server
Project-URL: Repository, https://github.com/HSPK/pp_rpc_server
Project-URL: Issues, https://github.com/HSPK/pp_rpc_server/issues
Project-URL: Upstream, https://github.com/earendil-works/pi/tree/main/packages/server
Author: Earendil Works
Maintainer-email: Hangxing Wei <HSPK@users.noreply.github.com>
License-Expression: MIT
License-File: LICENSE
Keywords: cbor,pi,protocol,rpc,server
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: pp-agent-core==0.1.0
Requires-Dist: pp-ai==0.1.0
Requires-Dist: pp-rpc-protocol==0.1.0
Description-Content-Type: text/markdown

# pi-server

Experimental. This package is under active development and may change or be removed without notice. Its APIs and behavior are not yet stable.

Server package for pi.

## Session server core

The package exports the `PiServer` session server.

```python
from typing import Any

from pi_server import CreateSessionOptions, PiServerService
from pi_server.transports.unix import UnixServerOptions, create_unix_server


class Service(PiServerService):
    async def list_sessions(self) -> list[dict[str, Any]]:
        return []

    async def list_models(self) -> list[dict[str, Any]]:
        return []

    async def create_session(self, options: CreateSessionOptions):
        raise NotImplementedError

    async def open_session(self, session_id: str):
        raise NotImplementedError


async def main() -> None:
    service = Service()
    server = create_unix_server(service, UnixServerOptions(path=".scratch/pi-server.sock"))
    await server.start()
```

`PiServer` composes transport listeners through the `PiServerListener` protocol. Each listener must complete any transport-specific authentication and authorization before passing a connection to `PiServer`. For example, a WebSocket listener can validate credentials during the HTTP upgrade, while the Unix listener relies on socket filesystem permissions. The Unix submodule exports the `create_unix_listener()` building block and `create_unix_server()` preset, keeping the common case concise without coupling the primary server to Unix sockets. The listener uses length-prefixed CBOR messages from `pi_protocol`.

This package does not provide a standalone CLI or coding-agent service. Applications supply the `PiServerService` implementation.

`PiServerService.list_sessions()` returns protocol `SessionMetadata`, not acquired runtime state. Services should map the durable fields their storage supports and may omit `updatedAt`, `parentSessionId`, `sessionName`, and `cwd`. `PiServer` refreshes available metadata from live snapshots without requiring stored sessions to fabricate phase, model, thinking-level, attachment, or lock values.

The real coding-agent service boundary lives in `pi_coding_agent.core.agent_session_runtime`. `pi_server` itself does not depend on `pi_coding_agent`.

## Transport testing

Custom transports can use `pi_server.testing` for deterministic protocol conformance tests. It exports `create_test_server()`, `TestServerService`, `TestSessionRuntime`, `ProtocolTestClient`, and the transport-neutral `WireChannel` contract. `connect_unix_test_client()` is provided for Unix transport tests.

```python
from pi_server import PiServerOptions
from pi_server.testing import TestServerService, create_test_server


async def main() -> None:
    service = TestServerService()
    harness = create_test_server(PiServerOptions(listeners=[]), service)
    await harness.server.start()
    await harness.server.close()
```

## `pi-ai` protocol bridge

`pi_ai` domain objects and `pi_protocol` wire DTOs remain independent. This package owns their boundary and exports `to_protocol_model_metadata()`, `to_protocol_assistant_message()`, `to_protocol_user_message()`, and `to_protocol_tool_result_message()`.

The adapters reject invalid tool inputs, identifiers, timestamps, and mismatched tool results; `to_protocol_tool_result_message()` requires the original `ToolCall` so it can verify the association and convert its arguments itself. Diagnostic details are explicitly sanitized. Closed `pi_ai` unions are mapped exhaustively. The protocol mirrors `pi_ai` vocabulary such as `toolCall` and `toolUse` where the semantics are identical. Protocol schemas enforce consistent lifecycle states, and tests encode adapter output through the runtime schemas so incompatible changes fail in the bridging package.

## Development

From the repository root:

```bash
uv sync --all-packages
uv run pytest packages/pi-server
uv run ruff check packages/pi-server
```

---

`pp-rpc-server` is developed in [HSPK/pp_rpc_server](https://github.com/HSPK/pp_rpc_server). It was split out of the `pp` monorepo; sibling packages (`pp-ai`, `pp-agent-core`, `pp-tui`, `pp-coding-agent`, ...) each live in their own
repository and are consumed from PyPI.
