Metadata-Version: 2.4
Name: genai-protocol
Version: 1.0.9
Summary: A universal agents connector library that integrates with GenAI.works infrastructure
Home-page: https://github.com/genai-works-org/genai-protocol
Author: Yaroslav Oliinyk, Valentyn Slivko, Ivan Kuzlo, Pavlo Shyrinskykh
Author-email: pavlo.shyrinskykh@genai.works
License: Apache License 2.0
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohappyeyeballs>=2.6.1
Requires-Dist: aiohttp>=3.11.16
Requires-Dist: aiosignal>=1.3.2
Requires-Dist: annotated-types>=0.7.0
Requires-Dist: attrs>=25.3.0
Requires-Dist: frozenlist>=1.5.0
Requires-Dist: idna>=3.10
Requires-Dist: multidict>=6.2.0
Requires-Dist: propcache>=0.3.1
Requires-Dist: pydantic>=2.11.1
Requires-Dist: pydantic-core>=2.33.0
Requires-Dist: typing-extensions>=4.13.0
Requires-Dist: typing-inspection>=0.4.0
Requires-Dist: websockets>=15.0.1
Requires-Dist: yarl>=1.18.3
Requires-Dist: pyjwt>=2.10.1
Provides-Extra: dev
Requires-Dist: twine>=6.1.0; extra == "dev"
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: license
Dynamic: license-file
Dynamic: provides-extra
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# GenAI Agent Protocol

## GenAI Agent Protocol is an async‑first Python framework for building WebSocket‑based AI agents that let you: 

* Connect an agent to the GenAI.works ecosystem 
* Process messages via registered handler functions
* Upload and retrieve files with contextual metadata (`agent_context`)
* Log messages with contextual metadata (`agent_context`)

## ✨ Features

🧠 Agent Binding: Decorator-based agent registration\
🪝 WebSocket Communication: Bidirectional messaging with a central server\
📁 File Manager: Async file upload/download & metadata fetch\
🪵 Context Logger: Structured, contextual WebSocket-based logging\
🔎 OpenAI Schema Conversion: Automatically converts Pydantic-type function signatures to OpenAI-compatible schemas\
📞 Agent‑to‑agent calls: invoke another registered agent from within your handler

## 📚 Core Concepts

**GenAISession**

A central controller that registers agents and manages the event lifecycle. 
```python
from genai_session.session import GenAISession

genai_session = GenAISession(
    jwt_token="<jwt received from GenAI CLI>",
    ws_url="<Router Container Public URL>", # Note if exposing infrastructure with ngrok
    api_base_url="<Backend Container Public URL>", # Note if exposing infrastructure with ngrok
)
```

**@bind(...)**

Registers a handler function with the session and make them visible to GenAI infrastructure.
```python
from genai_session.session import GenAISession
from genai_session.utils.context import GenAIContext

genai_session = GenAISession(jwt_token="<jwt received from GenAI CLI>")

@genai_session.bind(name="Test Name", description="Test Description")
async def message_handler(agent_context: GenAIContext, parameter: str) -> str:
    ...
```

**GenAIContext**

Provides contextual info (agent_uuid, request_id, etc.), a logger, and access to the FileManager.
```python
from genai_session.session import GenAISession
from genai_session.utils.context import GenAIContext

genai_session = GenAISession(jwt_token="<jwt received from GenAI CLI>")

@genai_session.bind(name="Test Name", description="Test Description")
async def message_handler(agent_context: GenAIContext, parameter: str) -> str:
    request_id = agent_context.request_id
```

**Files**

Handles file uploads (save) and retrievals (get_by_id, get_metadata_by_id).
```python
from genai_session.session import GenAISession
from genai_session.utils.context import GenAIContext

genai_session = GenAISession(jwt_token="<jwt received from GenAI CLI>")

@genai_session.bind(name="txt_content_reader_agent", description="Agent returns txt file content")
async def get_file_content(agent_context: GenAIContext, file_id: str) -> str:
    file = await agent_context.files.get_by_id(file_id)
    file_metadata = await agent_context.files.get_metadata_by_id(file_id)
    ...
```

**Logger**

Sends JSON logs through WebSocket with severity levels (debug, info, warning, error, critical).
```python
from genai_session.session import GenAISession
from genai_session.utils.context import GenAIContext

genai_session = GenAISession(jwt_token="<jwt received from GenAI CLI>")

@genai_session.bind()
async def reverse_name(agent_context: GenAIContext, name: str) -> str:
    """Agent reverses the name"""
    agent_context.logger.info("Inside the reverse_name function")
    agent_context.logger.debug(f"name: {name}")
    ...
```

**Invoke Agent from Agent**

You can invoke another agent from within an agent using the `genai_session.send` method.\
This method takes the `agent_uuid` and `params` as arguments.
```python
from genai_session.session import GenAISession
from genai_session.utils.context import GenAIContext
from genai_session.utils.agents import AgentResponse

genai_session = GenAISession(jwt_token="<jwt received from GenAI CLI>")

@genai_session.bind()
async def invoke_another_agent(agent_context: GenAIContext, name: dict) -> str:
    """Agent invokes another registered agent"""
    agent_response: AgentResponse = await genai_session.send(
        agent_uuid="agent_uuid", # you can get UUID from - await agent_context.get_my_agents()
        params={
            "username": name,
            "interests": ["python", "genai"],
            "age": 30,
        } # key is a parameter name, value is the value you want to pass
    )
    response = agent_response.response
    is_success = agent_response.is_success
    ...
```

## 📝 Function annotation examples

**No parameters**
```python
@genai_session.bind(name="GetCurrentDate", description="Return current date")
async def get_current_date(agent_context: GenAIContext):
    ...
```

**Built-in types**
```python
@genai_session.bind(name="Saver", description="Saves file")
async def file_saver(
    agent_context: GenAIContext,
    filename: str,
    file_content: str, 
    page_count: int, 
    images_names: list[str]
) -> dict:
    ...
```

**Pydantic models**
```python
from pydantic import BaseModel, Field
from typing import List, Any


class TranslationInput(BaseModel):
    text: str = Field(..., description="Text to translate")
    language: str = Field(..., description="Code of the language to translate to (e.g. 'fr', 'es')")
    banned_words: List[str] = Field(..., description="List of words to be banned from translation")

@genai_session.bind(name="TranslationAgent", description="Translate the text into specified language")
async def get_translation(
    agent_context: GenAIContext,
    params: TranslationInput
) -> dict[str, Any]:
    text = params.text
    language = params.language
    banned_words = params.banned_words
    ...
```

**typing Annotations**
```python
from typing import Any, Annotated

@genai_session.bind(name="TranslationAgent", description="Translate the text into specified language")
async def get_translation(
    agent_context: GenAIContext, 
    text: Annotated[str, "Text to translate"],
    language: Annotated[str, "Code of the language to translate to (e.g. 'fr', 'es')"],
    banned_words: Annotated[list[str], "List of words to be banned from translation"],
) -> dict[str, Any]:
    ...
```
