reminix.runtime.Agent¶

class reminix.runtime.Agent[source]¶

Bases: object

An AI agent that can handle invoke and chat requests.

Use separate handlers for non-streaming and streaming responses: - @agent.invoke / @agent.chat for non-streaming - @agent.invoke_stream / @agent.chat_stream for streaming

Example:

agent = Agent(name="my-agent")

@agent.invoke
async def handle_invoke(input: dict, ctx: Context):
    return {"output": f"Processed: {input}"}

@agent.invoke_stream
async def handle_invoke_stream(input: dict, ctx: Context):
    for word in ["Hello", " ", "World"]:
        yield {"chunk": word}

@agent.chat
async def handle_chat(messages: list, ctx: Context):
    return {"message": {"role": "assistant", "content": "Hello!"}}

@agent.chat_stream
async def handle_chat_stream(messages: list, ctx: Context):
    for word in ["Hello", "!", " How can I help?"]:
        yield {"chunk": word}

__init__(name, *, metadata=None)[source]¶

Initialize an agent with a name and optional metadata.

Parameters:

name (str) – Unique name for the agent (used in URL routing). Must be URL-safe (alphanumeric, hyphens, underscores).
metadata (dict[str, Any] | None) – Optional user-defined metadata (e.g., framework, model, description).

Return type:

None

Methods

`__init__`(name, *[, metadata])	Initialize an agent with a name and optional metadata.
`chat`(handler)	Decorator to register a non-streaming chat handler.
`chat_stream`(handler)	Decorator to register a streaming chat handler.
`handle_chat`(messages[, ctx])	Execute the non-streaming chat handler.
`handle_chat_stream`(messages[, ctx])	Execute the streaming chat handler.
`handle_invoke`(input_data[, ctx])	Execute the non-streaming invoke handler.
`handle_invoke_stream`(input_data[, ctx])	Execute the streaming invoke handler.
`invoke`(handler)	Decorator to register a non-streaming invoke handler.
`invoke_stream`(handler)	Decorator to register a streaming invoke handler.
`to_info`()	Return agent information for discovery.

Attributes

`has_chat`	Check if non-streaming chat handler is registered.
`has_chat_stream`	Check if streaming chat handler is registered.
`has_invoke`	Check if non-streaming invoke handler is registered.
`has_invoke_stream`	Check if streaming invoke handler is registered.
`metadata`	Get the agent's metadata.

__init__(name, *, metadata=None)[source]¶

Initialize an agent with a name and optional metadata.

Parameters:

name (str) – Unique name for the agent (used in URL routing). Must be URL-safe (alphanumeric, hyphens, underscores).
metadata (dict[str, Any] | None) – Optional user-defined metadata (e.g., framework, model, description).

Return type:

None

property metadata: dict[str, Any]¶: Get the agent’s metadata.

to_info()[source]¶

Return agent information for discovery.

Return type:: AgentInfo
Returns:: AgentInfo dict with name, capabilities, and metadata.

invoke(handler)[source]¶

Decorator to register a non-streaming invoke handler.

The handler receives the input dict and should return {“output”: …}.

Example:

@agent.invoke
async def handle_invoke(input: dict):
    return {"output": f"Result: {input['query']}"}

Return type:: Callable[[dict[str, Any]], Any]
Parameters:: handler (Callable[[dict[str, Any]], Any])

invoke_stream(handler)[source]¶

Decorator to register a streaming invoke handler.

The handler receives the input dict and should yield {“chunk”: “…”}.

Example:

@agent.invoke_stream
async def handle_invoke_stream(input: dict):
    for word in ["Hello", " ", "World"]:
        yield {"chunk": word}

Return type:: Callable[[dict[str, Any]], Any]
Parameters:: handler (Callable[[dict[str, Any]], Any])

chat(handler)[source]¶

Decorator to register a non-streaming chat handler.

The handler receives the messages list and should return {“message”: {“role”: “assistant”, “content”: “…”}}.

Example:

@agent.chat
async def handle_chat(messages: list):
    return {
        "message": {
            "role": "assistant",
            "content": "Hello! How can I help?"
        }
    }

Return type:: Callable[[list[ChatMessage]], Any]
Parameters:: handler (Callable[[list[ChatMessage]], Any])

chat_stream(handler)[source]¶

Decorator to register a streaming chat handler.

The handler receives the messages list and should yield {“chunk”: “…”}.

Example:

@agent.chat_stream
async def handle_chat_stream(messages: list):
    for word in ["Hello", "!", " How can I help?"]:
        yield {"chunk": word}

Return type:: Callable[[list[ChatMessage]], Any]
Parameters:: handler (Callable[[list[ChatMessage]], Any])

property has_invoke: bool¶: Check if non-streaming invoke handler is registered.

property has_invoke_stream: bool¶: Check if streaming invoke handler is registered.

property has_chat: bool¶: Check if non-streaming chat handler is registered.

property has_chat_stream: bool¶: Check if streaming chat handler is registered.

async handle_invoke(input_data, ctx=None)[source]¶

Execute the non-streaming invoke handler.

Parameters:

input_data (dict[str, Any]) – The input data from the request.
ctx (Context | None) – Optional request context (passed if handler accepts it).

Return type:

InvokeResponse

Returns:

An InvokeResponse dict with output.

Raises:

NotImplementedError – If no invoke handler is registered.

async handle_invoke_stream(input_data, ctx=None)[source]¶

Execute the streaming invoke handler.

Parameters:

input_data (dict[str, Any]) – The input data from the request.
ctx (Context | None) – Optional request context (passed if handler accepts it).

Return type:

AsyncGenerator[StreamChunk, None]

Returns:

An async generator yielding StreamChunks.

Raises:

NotImplementedError – If no streaming invoke handler is registered.

async handle_chat(messages, ctx=None)[source]¶

Execute the non-streaming chat handler.

Parameters:

messages (list[ChatMessage]) – The chat messages from the request.
ctx (Context | None) – Optional request context (passed if handler accepts it).

Return type:

ChatResponse

Returns:

A ChatResponse dict with message.

Raises:

NotImplementedError – If no chat handler is registered.

async handle_chat_stream(messages, ctx=None)[source]¶

Execute the streaming chat handler.

Parameters:

messages (list[ChatMessage]) – The chat messages from the request.
ctx (Context | None) – Optional request context (passed if handler accepts it).

Return type:

AsyncGenerator[StreamChunk, None]

Returns:

An async generator yielding StreamChunks.

Raises:

NotImplementedError – If no streaming chat handler is registered.