fjagepy 2.1.2

fjage Gateway in Python

fjåge Gateway in Python (fjagepy)

fjagepy is a Python gateway for the fjåge multi-agent system framework. It enables Python applications to seamlessly communicate with fjåge agents over TCP connections, providing a Pythonic interface to the fjåge ecosystem.

Features

• Agent Communication: Send and receive messages to/from fjåge agents
• Request-Response Pattern: Built-in support for synchronous request-response interactions
• Message Filtering: Flexible message filtering by type, sender, or custom predicates
• Agent Discovery: Find and communicate with agents by name or service
• NumPy Integration: Automatic serialization/deserialization of NumPy arrays
• Connection Management: Automatic reconnection and robust error handling
• Type Safety: Full type hints for better IDE support and code reliability

Installation

From PyPI

pip install fjagepy

From Source

git clone https://github.com/org-arl/fjage.git
cd fjage/gateways/python
pip install -e .

Development Installation

For development with testing dependencies:

pip install -e ".[dev]"

Quick Start

Basic Connection and Communication

from fjagepy import Gateway, Message, AgentID

# Connect to fjåge container
gw = Gateway('localhost', 1100)

# Get reference to an agent
shell = gw.agent('shell')

# Send a message
msg = Message()
msg.data = "Hello from Python!"
shell << msg

# Close connection when done
gw.close()

Request-Response Pattern

from fjagepy import Gateway, ShellExecReq

# Connect and get shell agent
gw = Gateway()
shell = gw.agent('shell')

# Create and send a shell command request
req = ShellExecReq()
req.command = 'ps'
req.ans = True

# Send request and wait for response
rsp = shell.request(req, timeout=5000)  # 5 second timeout
if rsp:
    print(f"Command output: {rsp.output}")

gw.close()

Message Filtering and Reception

from fjagepy import Gateway, Message

gw = Gateway()

# Receive any message
msg = gw.receive(timeout=1000)

# Receive messages of specific type
from fjagepy import ParameterRsp
param_msg = gw.receive(ParameterRsp, timeout=1000)

# Receive with custom filter
def my_filter(msg):
    return hasattr(msg, 'data') and 'important' in str(msg.data)

important_msg = gw.receive(my_filter, timeout=1000)

gw.close()

User-Defined Message Classes

Use @message to register your own message subclasses so incoming JSON can be inflated back into the right Python type.

from fjagepy import Message, message


@message('org.example.CustomReq')
class CustomReq(Message):
    def __init__(self, text=None):
        super().__init__()
        self.text = text


outgoing = CustomReq('hello')
payload = outgoing.to_json()

# Later, when a message with the same clazz arrives from fjage:
incoming = Message.from_json(payload)
assert isinstance(incoming, CustomReq)
assert incoming.text == 'hello'

Advanced Usage

Custom Connectors

Implement custom connectors by extending the Connector base class:

from fjagepy import Connector

class MyCustomConnector(Connector):
    def connect(self):
        # Implementation
        pass

    def disconnect(self):
        # Implementation
        pass

    def send(self, msg: str):
        # Implementation
        pass

# Use custom connector
gw = Gateway(connector=MyCustomConnector)

Logging and Debugging

This library uses the standard Python logging system.

By default, the library does not emit logs to the console. To see logs, configure Python’s logging in your application:

import logging

# Show all logs on stdout
logging.basicConfig(level=logging.DEBUG)

# Or, enable only logs from this library
logging.getLogger("fjagepy").setLevel(logging.DEBUG)

For troubleshooting, you can also send logs to a file:

logging.basicConfig(filename="debug.log", level=logging.DEBUG)

Context Manager Support

Use Gateway as a context manager for automatic cleanup:

from fjagepy import Gateway

with Gateway('localhost', 5081) as gw:
    shell = gw.agent('shell')
    req = ShellExecReq()
    req.command = 'help'
    req.ans = True
    rsp = shell.request(req)
    print(rsp.output)
# Gateway automatically closed

iPython / Jupyter Notebook Support

In Jupyter notebooks, AgentIDs and Messages are displayed in a human-readable format

In [1]: from fjagepy import *

In [2]: gw = Gateway("localhost", 5081)

In [3]: gw.agentForService(Services.SHELL)
Out[3]:
<<< websh >>>

Interactive Groovy shell

[org.arl.fjage.shell.ShellParam]
  language => Groovy

Testing

Run the test suite:

# Install test dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run specific test file
pytest test/test_gateway.py

# Run with coverage
pytest --cov=fjagepy

Test Requirements

Tests require a running fjåge container on localhost:5081. You can start one using:

# In the main fjåge directory
./gradlew --info -PmanualPyTest=true test --tests="org.arl.fjage.test.fjagepyTest"

Examples

See the examples/ directory an example of how to use the fjagepy gateway.

Documentation

To build the documentation locally:

# Install documentation dependencies
pip install -e ".[docs]"
# Build docs
sphinx-build -b html docs/ ../../docs/pydocs/
# View in browser
open ../../docs/pydocs/index.html

Compatibility

• Python: 3.9 or higher
• fjåge: Compatible with fjåge 1.7.x and 2.x
• NumPy: Optional dependency for array support
• Operating Systems: Linux, macOS, Windows

Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Add tests for new functionality
5. Run the test suite (pytest)
6. Commit your changes (git commit -m 'Add amazing feature')
7. Push to the branch (git push origin feature/amazing-feature)
8. Open a Pull Request

License

This project is licensed under the BSD 3-Clause License. See the LICENSE file for details.

Support

• Documentation: fjåge Documentation
• Issues: GitHub Issues
• Discussions: GitHub Discussions

Related Projects

• fjåge: Main fjåge framework
• fjåge.js: JavaScript gateway
• Fjåge.jl: Julia gateway
• fjåge-c: C gateway