grpchook 0.0.5

grpchook (grpc + hook) is an asynchronous Python gRPC bidirectional-streaming framework. Subclass BaseServer/BaseClient, override hooks — the base handles all gRPC plumbing.

grpchook

grpchook (gRPC + hook) is a Python framework for building asynchronous gRPC bidirectional-streaming services. Subclass BaseServer and BaseClient, override the hooks you need — the framework handles all gRPC plumbing.

Status: Work in Progress. The project is open source and will remain open source. Treat with caution. If you depend on it, pin your version.

---

Table of Contents

• Disclaimer
• When to Use and When Not to Use grpchook
• Requirements
• Installation
  • From PyPI
  • From Source
• Quick Start
• Parameters
• Minimal Examples
• Examples
• Testing
• Extend default Configuration
• Regenerating the gRPC Interface
• ToDos and Roadmap
• Known Issues and Troubleshooting
• Contributing
• License
• Release History

---

Disclaimer

Core concept by a human developer. AI was used to assist with unit and integration tests, examples, documentation, and selected code sections. Core logic has been reviewed by a human; the test suite has not been fully audited — there may be AI-introduced oversights. Please report any issues you find.

This software is provided "as is", without warranty of any kind. The developer is not responsible for any damage, data loss, security vulnerabilities, or other issues that may arise from using this software. You use it at your own risk. See LICENSE.txt for the full BSD 3-Clause terms.

---

When to Use and When Not to Use grpchook

When to Use grpchook

• You need a simple, Python-based gRPC bidirectional streaming server and client.

• You want a data exchange blueprint for developers or AI agents to build on top of.

• You want a framework that can be extended with custom hooks for specific events.

• You want to distribute clients to many different machines (e.g. voice recorder, voice to text, text to LLM, and vice versa until the final response is replayed)

  Example — four clients on four machines, all routed through one grpchook server:

  💡 Diagram requires the Markdown Preview Mermaid Support extension to render in VS Code.

  flowchart LR
      subgraph M1["📦 Machine 1"]
          VR["🎤 Voice Recorder"]
      end
      subgraph M2["📦 Machine 2"]
          STT["📝 Speech-to-Text"]
      end
      subgraph M3["📦 Machine 3"]
          LLM["🤖 LLM Processor"]
      end
      subgraph M4["📦 Machine 4"]
          RP["🔊 Voice Replay"]
      end
  
      SRV(["⚙️ grpchook Server"])
  
      VR  -->|"① audio"| SRV
      SRV -->|"① audio"| STT
      STT -->|"② transcript"| SRV
      SRV -->|"② transcript"| LLM
      LLM -->|"③ llm_response"| SRV
      SRV -->|"③ llm_response"| RP
  

When Not to Use grpchook

• When you need a very large number of clients; the threading model may introduce overhead.
• When you need direct peer-to-peer communication without a server intermediary; grpchook routes all messages through a central server.
• You want a framework that supports multiple programming languages out of the box; grpchook is (currently) Python-only.

---

Requirements

• Python 3.10 or later
• A dedicated virtual environment is strongly recommended — gRPC version conflicts with other packages are common when using grpchook.

---

Installation

From PyPI

pip install grpchook

From Source

git clone https://github.com/fwkrumm/grpchook.git
cd grpchook
pip install -e .

---

Quick Start

Refer to HOW_TO.md for the full API reference and code examples. Alternatively run

python -m grpchook --generate-skeletons

to generate a very basic server and client skeleton in the current directory. Use

python -m grpchook --generate-interface-with-skeletons

to generate the skeletons along with a copy of the message.proto interface file in the current directory to modify which is then used by the skeletons.

---

Parameters

You can print the following text via python -m grpchook --help:

usage: python -m grpchook [-h] [--generate] [--generate-skeletons] [--generate-server] [--generate-client] [--generate-how-to] [--generate-interface] [--generate-interface-with-skeletons]

grpchook scaffolding tool.

Generates skeleton server/client files and the HOW_TO reference
document into the current working directory.

options:
  -h, --help            show this help message and exit
  --generate            Generate server_skeleton.py, client_skeleton.py, and HOW_TO.md
  --generate-skeletons  Generate server_skeleton.py and client_skeleton.py
  --generate-server     Generate server_skeleton.py only
  --generate-client     Generate client_skeleton.py only
  --generate-how-to     Copy HOW_TO.md into the current directory
  --generate-interface  Copy message.proto into the current directory and print customisation instructions
  --generate-interface-with-skeletons
                        Copy message.proto and write server_skeleton.py + client_skeleton.py that load the custom interface at startup via compile_and_register()

examples:
  python -m grpchook --generate                          # skeleton + HOW_TO
  python -m grpchook --generate-skeletons                  # server + client only
  python -m grpchook --generate-server                   # server only
  python -m grpchook --generate-client                   # client only
  python -m grpchook --generate-how-to                   # HOW_TO.md only
  python -m grpchook --generate-interface                # message.proto + instructions
  python -m grpchook --generate-interface-with-skeletons  # proto + matching skeletons

---

Minimal Examples

Ultra-minimal — no subclassing required

The simplest possible working setup: start a server, connect two clients, exchange a message. Everything runs in a single script — no subclassing or hook overrides needed.

# example_minimal.py
import threading
from grpchook.baseserver import BaseServer
from grpchook.baseclient import BaseClient
from grpchook.tools import generate_message

# start the server in a background thread
server = BaseServer(port=50051, name="server")
threading.Thread(target=server.serve_forever, daemon=True).start()

# both clients declare the same channel name
# fan-out skips the sender, so client_b receives what client_a sends
client_a = BaseClient(port=50051, name="A", provides=["ping"], requires=["ping"])
client_b = BaseClient(port=50051, name="B", provides=["ping"], requires=["ping"])

client_a.send_data(generate_message("ping", byte_payload=b"hello"))

msg = client_b.get_data(timeout=5.0)
client_a.logger.info(msg.payload.bytePayload)   # b"hello"
client_b.logger.info(msg.payload.bytePayload)   # b"hello"

client_a.disconnect()
client_b.disconnect()
server.shutdown()

Request / response — subclass with hooks

For real workloads, subclass BaseServer to control routing and BaseClient to react to messages via the on_receive hook.

server.py

from grpchook.baseserver import BaseServer, Peer
from grpchook.tools import generate_message
import grpchook.message_pb2 as pb2


class EchoServer(BaseServer):
    def __init__(self):
        super().__init__(port=50051, name="echo-server")

    def on_receive(self, peer: Peer, request: pb2.Message) -> bool:
        if request.metaInfo.messageName == "request":
            reply = generate_message("response", byte_payload=request.payload.bytePayload)
            self._data_register.add_data_for_message_name(
                peer.client_id, "response", reply,
                target_client_id=peer.client_id,   # unicast back to sender
            )
            return False   # skip default fan-out; routing handled above
        return True


EchoServer().serve_forever()

client.py

from grpchook.baseclient import BaseClient
from grpchook.tools import generate_message
import grpchook.message_pb2 as pb2


class EchoClient(BaseClient):
    def __init__(self):
        super().__init__(port=50051, name="echo-client",
                         provides=["request"], requires=["response"])

    def on_receive(self, data: pb2.Message):
        print(f"Server replied: {data.payload.bytePayload.decode()}")


client = EchoClient()
client.send_data(generate_message("request", byte_payload=b"hello, grpchook!"))
client.spin(timeout=5.0)   # calls on_receive() per message; returns on timeout/disconnect
client.disconnect()

Run the server first, then the client:

# terminal 1
python server.py

# terminal 2
python client.py

---

Examples

Runnable examples are available in two locations:

• examples/ — self-contained, scenario-focused examples
• tests/integration/ — integration test scenarios covering a broad range of use cases

Run them on a machine with adequate resources; some scenarios are resource-intensive.

---

Testing

Install dev dependencies and run the unit tests:

pip install -r requirements_dev.txt
python -m unittest discover -s tests

Integration tests are in tests/integration/ and can be run via:

python tests/integration/run_integration_tests.py

---

Extend default Configuration

Example for a client to use the default configuration but disable proxy forwarding:

from grpchook.baseclient import BaseClient, ClientConfig

class TestClient(BaseClient):
    def __init__(self):
        config = ClientConfig()
        config.grpc_options += [("grpc.enable_http_proxy", 0)]
        super().__init__(port=50051, name="test-client", config=config)

---

Regenerating the gRPC Interface

If you modify grpchook/message.proto after cloning the repository, regenerate the Python bindings with:

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. --pyi_out=. grpchook/message.proto

Note that all clients which connect to a server have to use the same proto schema version i.e. the same proto file. The different signals for the clients must be used in substructures:

message Payload {
    // For client A
    SomeTypeA payloadClientA = 1;

	// For client B
    SomeTypeB payloadClientB = 2;

    ...
}

---

ToDos and Roadmap

Performance & Stability

• Evaluate replacing the threading model with asyncio if the performance gain justifies the API tradeoff.
• Verify behavior when connections are interrupted mid-stream; ensure no ghost threads or queue deadlocks occur.

Planned Features

• Multi-language client example (e.g., C++ or Rust).
• SSL/TLS usage example.

---

Known Issues and Troubleshooting

TBD

---

Contributing

Contributions are welcome. Please open an issue first for major changes so the approach can be discussed. For bug fixes and small improvements, a pull request is sufficient.

---

License

BSD 3-Clause — see LICENSE.txt.

---

Release History

Version / Git Tag on Master	Description
0.0.1	Unpublished.
0.0.2	Initial public release.
0.0.3	Add ms timestamp resolution to log output and minor adjustments to readme.
0.0.4	Add executor for server and wait for shutdown.
0.0.5	Fix readme on pypi page.