[Feature][Python] Implement EventListener to align with Java SDK

[twosom]

Search before asking

• I searched in the issues and found nothing similar.

Related PR

• [Feature][Java] Add java event listeners config #641
• [Feature] [api] Support Event Logger and Event Listener in Java #102

Description

This feature introduces the EventListener interface to enable event-driven monitoring and custom logic extensions within the Flink Agent Python SDK, aligning with the existing Java SDK implementation.

Design Considerations

During the initial design phase, I explored two different approaches for registering event listeners:

Option 1: Decorator-based Registration (Considered)
I initially evaluated a Pythonic approach where a decorator registers a standalone function as a listener by storing its metadata in an internal registry.

# Information stored in an internal registry
REGISTRY = []

def event_listener(func):
    REGISTRY.append({
        "module": func.__module__,
        "qualname": func.__qualname__
    })
    return func

@event_listener
def my_handler(context, event):
    print(f"Received: {event}")

Pros: Highly idiomatic and concise for Python developers.
Cons: This approach deviates significantly from the class/interface-based structure of the Java SDK, breaking API consistency across different language SDKs.

Option 2: Class-based Interface (Final Decision)
To maintain a unified developer experience and ensure structural alignment with the Java SDK, I decided to adopt a class-based interface design.

Proposed API Usage

The user implements the EventListener interface, and the framework dynamically instantiates and triggers it during the event lifecycle.

from flink_agents.api.listener import EventListener
from flink_agents.api.event_context import EventContext
from flink_agents.api.events import Event

class MyCustomListener(EventListener):
    def on_event_processed(self, context: EventContext, event: Event) -> None:
        print(f"Event processed: {event.type} at {context.timestamp}")

# Registration via environment configuration
agents_env.get_config().set(
    AgentConfigOptions.EVENT_LISTENERS, 
    [str(MyCustomListener)]
)

Are you willing to submit a PR?

• I'm willing to submit a PR!