| Name | adk-agui-py-middleware JSON |
| Version |
0.1.3
JSON |
| download |
| home_page | None |
| Summary | Python middleware bridging Google ADK agents with AGUI protocol via Server-Sent Events for real-time agent interactions |
| upload_time | 2025-08-13 07:36:22 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.10 |
| license | MIT License
Copyright (c) 2025 DennySORA
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE. |
| keywords |
agent
agui
fastapi
google-adk
middleware
sse
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# ADK AGUI Python Middleware
A professional Python middleware library that bridges Agent Development Kit (ADK) agents with AGUI (Agent UI) protocol, providing Server-Sent Events (SSE) streaming capabilities for real-time agent interactions.
## Overview
This middleware facilitates seamless integration between Google's Agent Development Kit and AGUI-compatible frontends, enabling developers to build interactive agent applications with real-time streaming responses. The library handles agent execution, session management, event encoding, and error handling in a production-ready framework.
## Features
- **Real-time Streaming**: Server-Sent Events (SSE) support for live agent responses
- **Session Management**: Comprehensive session handling with configurable backends
- **Context Extraction**: Flexible context configuration for multi-tenant applications
- **Error Handling**: Robust error handling with detailed logging and recovery
- **Tool Integration**: Support for tool calls and tool result processing
- **Event Encoding**: Multiple event encoding formats with automatic content negotiation
- **Type Safety**: Full type annotations with Pydantic models
- **Extensible Architecture**: Abstract base classes for custom implementations
## Architecture
The middleware follows a modular architecture with clear separation of concerns:
```
├── base_abc/ # Abstract base classes
├── config/ # Configuration and logging setup
├── data_model/ # Pydantic models and data structures
├── endpoint.py # FastAPI endpoint registration
├── event/ # Event handling and error events
├── handler/ # Request and session handlers
├── loggers/ # Logging infrastructure
├── manager/ # Session and resource management
├── sse_service.py # Core SSE service implementation
└── tools/ # Utility functions and converters
```
## Installation
### Using pip
```bash
pip install adk-agui-py-middleware
```
Available on PyPI: https://pypi.org/project/adk-agui-py-middleware/
### Requirements
- Python 3.13+
- Google ADK (`google-adk>=1.10.0`)
- AGUI Protocol (`ag-ui-protocol>=0.1.8`)
- Pydantic (`pydantic>=2.11.7`)
### Development Install
```bash
# Using uv (recommended)
uv sync
# Or using pip
pip install -r requirements.txt
```
## Quick Start
### Basic Setup
```python
from fastapi import FastAPI
from google.adk.agents import BaseAgent
from adk_agui_middleware import register_agui_endpoint, SSEService
from adk_agui_middleware.data_model.context import RunnerConfig, ContextConfig
# Create your FastAPI app
app = FastAPI()
# Initialize your ADK agent
agent = BaseAgent() # Your custom agent implementation
# Configure context extraction
context_config = ContextConfig(
app_name="my-agent-app",
user_id="user-123", # Can be a callable for dynamic extraction
)
# Configure runner services
runner_config = RunnerConfig(
use_in_memory_services=True # Use in-memory services for development
)
# Create SSE service
sse_service = SSEService(
agent=agent,
runner_config=runner_config,
context_config=context_config
)
# Register the AGUI endpoint
register_agui_endpoint(app, sse_service, path="/agui")
```
### Dynamic Context Extraction
For multi-tenant applications, you can extract context dynamically from requests:
```python
from ag_ui.core import RunAgentInput
from fastapi import Request
async def extract_user_id(agui_content: RunAgentInput, request: Request) -> str:
"""Extract user ID from JWT token or headers"""
token = request.headers.get("authorization")
# Your authentication logic here
return decoded_user_id
async def extract_app_name(agui_content: RunAgentInput, request: Request) -> str:
"""Extract app name from subdomain or headers"""
host = request.headers.get("host", "")
return host.split(".")[0] if "." in host else "default"
context_config = ContextConfig(
app_name=extract_app_name,
user_id=extract_user_id,
session_id=lambda content, req: content.thread_id # Use thread ID as session
)
```
### Custom Services Configuration
For production deployments, configure external services:
```python
from google.adk.sessions import DatabaseSessionService
from google.adk.memory import RedisMemoryService
runner_config = RunnerConfig(
use_in_memory_services=False,
session_service=DatabaseSessionService(connection_string="..."),
memory_service=RedisMemoryService(redis_url="..."),
# artifact_service and credential_service as needed
)
```
## Advanced Usage
### Custom SSE Service
Extend the base SSE service for custom behavior:
```python
from adk_agui_middleware.base_abc.sse_service import BaseSSEService
class CustomSSEService(BaseSSEService):
async def get_runner(self, agui_content, request):
# Custom runner logic
pass
async def event_generator(self, runner, encoder):
# Custom event generation
async for event in runner():
# Custom event processing
yield encoder.encode(event)
```
### Error Handling
The middleware includes comprehensive error handling with logging:
```python
from adk_agui_middleware.loggers.logger import setup_logging
# Configure logging
setup_logging(level="INFO")
# Error events are automatically generated for:
# - Agent execution errors
# - Encoding failures
# - Tool execution errors
# - Session management issues
```
### Tool Integration
Handle tool calls and results seamlessly:
```python
# Tool calls are automatically extracted from AGUI messages
# Tool results are processed and converted to the appropriate format
# No additional configuration required - handled by UserMessageHandler
```
## API Reference
### Core Classes
- **`SSEService`**: Main service implementation for handling agent interactions
- **`register_agui_endpoint()`**: Function to register the AGUI endpoint on FastAPI
- **`ContextConfig`**: Configuration for extracting context from requests
- **`RunnerConfig`**: Configuration for ADK runner and services
- **`UserMessageHandler`**: Processes user messages and tool results
### Data Models
- **`SessionParameter`**: Session identification parameters
- **`BaseEvent`**: Base class for agent events
- **Various error models**: Structured error handling
## Development
### Code Quality
The project uses comprehensive linting and formatting:
```bash
# Format code
ruff format
# Lint code
ruff check
# Type checking
mypy src/
```
### Testing
```bash
# Run tests (configure based on your test setup)
pytest
# With coverage
pytest --cov=adk_agui_middleware
```
## Configuration
### Environment Variables
Configure the middleware using environment variables or configuration files:
```bash
# Logging level
LOG_LEVEL=INFO
# Service configurations
USE_IN_MEMORY_SERVICES=true
SESSION_SERVICE_URL=redis://localhost:6379
```
### Runtime Configuration
```python
from adk_agui_middleware.config.log import setup_logging
# Configure logging
setup_logging(
level="DEBUG",
format="json", # or "text"
output="file" # or "console"
)
```
## Production Considerations
### Performance
- Use external services (Redis, PostgreSQL) instead of in-memory services
- Configure appropriate connection pooling
- Monitor memory usage for long-running sessions
- Implement proper session cleanup policies
### Security
- Validate all incoming requests
- Implement proper authentication and authorization
- Use HTTPS in production
- Configure CORS appropriately
- Sanitize user inputs and tool results
### Monitoring
- Monitor agent execution times
- Track error rates and types
- Monitor session lifecycle
- Log security events and access patterns
## Contributing
1. Follow the existing code style and conventions
2. Add comprehensive docstrings using Google style
3. Include type annotations for all functions
4. Write tests for new functionality
5. Update documentation as needed
### Acknowledgments
This project is inspired by and builds upon the foundational work of [@contextablemark](https://github.com/contextablemark) and their [AG-UI implementation](https://github.com/contextablemark/ag-ui). We extend our gratitude for their innovative approach to agent-UI integration, which provided essential architectural insights and design patterns that shaped this middleware development.
## License
This project is licensed under the terms specified in the LICENSE file.
## Support
For issues and questions:
1. Check the documentation and examples
2. Review existing issues in the repository
3. Create a new issue with detailed information about your use case
Raw data
{
"_id": null,
"home_page": null,
"name": "adk-agui-py-middleware",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": "DennySORA <dennysora.main@gmail.com>",
"keywords": "agent, agui, fastapi, google-adk, middleware, sse",
"author": null,
"author_email": "DennySORA <dennysora.main@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/da/e4/02b221ba506a0aa548205d036a0940b51d7bb7578bf9622416e576c03184/adk_agui_py_middleware-0.1.3.tar.gz",
"platform": null,
"description": "# ADK AGUI Python Middleware\n\nA professional Python middleware library that bridges Agent Development Kit (ADK) agents with AGUI (Agent UI) protocol, providing Server-Sent Events (SSE) streaming capabilities for real-time agent interactions.\n\n## Overview\n\nThis middleware facilitates seamless integration between Google's Agent Development Kit and AGUI-compatible frontends, enabling developers to build interactive agent applications with real-time streaming responses. The library handles agent execution, session management, event encoding, and error handling in a production-ready framework.\n\n## Features\n\n- **Real-time Streaming**: Server-Sent Events (SSE) support for live agent responses\n- **Session Management**: Comprehensive session handling with configurable backends\n- **Context Extraction**: Flexible context configuration for multi-tenant applications \n- **Error Handling**: Robust error handling with detailed logging and recovery\n- **Tool Integration**: Support for tool calls and tool result processing\n- **Event Encoding**: Multiple event encoding formats with automatic content negotiation\n- **Type Safety**: Full type annotations with Pydantic models\n- **Extensible Architecture**: Abstract base classes for custom implementations\n\n## Architecture\n\nThe middleware follows a modular architecture with clear separation of concerns:\n\n```\n\u251c\u2500\u2500 base_abc/ # Abstract base classes\n\u251c\u2500\u2500 config/ # Configuration and logging setup\n\u251c\u2500\u2500 data_model/ # Pydantic models and data structures\n\u251c\u2500\u2500 endpoint.py # FastAPI endpoint registration\n\u251c\u2500\u2500 event/ # Event handling and error events\n\u251c\u2500\u2500 handler/ # Request and session handlers\n\u251c\u2500\u2500 loggers/ # Logging infrastructure\n\u251c\u2500\u2500 manager/ # Session and resource management\n\u251c\u2500\u2500 sse_service.py # Core SSE service implementation\n\u2514\u2500\u2500 tools/ # Utility functions and converters\n```\n\n## Installation\n\n### Using pip\n\n```bash\npip install adk-agui-py-middleware\n```\n\nAvailable on PyPI: https://pypi.org/project/adk-agui-py-middleware/\n\n### Requirements\n\n- Python 3.13+\n- Google ADK (`google-adk>=1.10.0`)\n- AGUI Protocol (`ag-ui-protocol>=0.1.8`)\n- Pydantic (`pydantic>=2.11.7`)\n\n### Development Install\n\n```bash\n# Using uv (recommended)\nuv sync\n\n# Or using pip\npip install -r requirements.txt\n```\n\n## Quick Start\n\n### Basic Setup\n\n```python\nfrom fastapi import FastAPI\nfrom google.adk.agents import BaseAgent\nfrom adk_agui_middleware import register_agui_endpoint, SSEService\nfrom adk_agui_middleware.data_model.context import RunnerConfig, ContextConfig\n\n# Create your FastAPI app\napp = FastAPI()\n\n# Initialize your ADK agent\nagent = BaseAgent() # Your custom agent implementation\n\n# Configure context extraction\ncontext_config = ContextConfig(\n app_name=\"my-agent-app\",\n user_id=\"user-123\", # Can be a callable for dynamic extraction\n)\n\n# Configure runner services\nrunner_config = RunnerConfig(\n use_in_memory_services=True # Use in-memory services for development\n)\n\n# Create SSE service\nsse_service = SSEService(\n agent=agent,\n runner_config=runner_config,\n context_config=context_config\n)\n\n# Register the AGUI endpoint\nregister_agui_endpoint(app, sse_service, path=\"/agui\")\n```\n\n### Dynamic Context Extraction\n\nFor multi-tenant applications, you can extract context dynamically from requests:\n\n```python\nfrom ag_ui.core import RunAgentInput\nfrom fastapi import Request\n\nasync def extract_user_id(agui_content: RunAgentInput, request: Request) -> str:\n \"\"\"Extract user ID from JWT token or headers\"\"\"\n token = request.headers.get(\"authorization\")\n # Your authentication logic here\n return decoded_user_id\n\nasync def extract_app_name(agui_content: RunAgentInput, request: Request) -> str:\n \"\"\"Extract app name from subdomain or headers\"\"\"\n host = request.headers.get(\"host\", \"\")\n return host.split(\".\")[0] if \".\" in host else \"default\"\n\ncontext_config = ContextConfig(\n app_name=extract_app_name,\n user_id=extract_user_id,\n session_id=lambda content, req: content.thread_id # Use thread ID as session\n)\n```\n\n### Custom Services Configuration\n\nFor production deployments, configure external services:\n\n```python\nfrom google.adk.sessions import DatabaseSessionService\nfrom google.adk.memory import RedisMemoryService\n\nrunner_config = RunnerConfig(\n use_in_memory_services=False,\n session_service=DatabaseSessionService(connection_string=\"...\"),\n memory_service=RedisMemoryService(redis_url=\"...\"),\n # artifact_service and credential_service as needed\n)\n```\n\n## Advanced Usage\n\n### Custom SSE Service\n\nExtend the base SSE service for custom behavior:\n\n```python\nfrom adk_agui_middleware.base_abc.sse_service import BaseSSEService\n\nclass CustomSSEService(BaseSSEService):\n async def get_runner(self, agui_content, request):\n # Custom runner logic\n pass\n \n async def event_generator(self, runner, encoder):\n # Custom event generation\n async for event in runner():\n # Custom event processing\n yield encoder.encode(event)\n```\n\n### Error Handling\n\nThe middleware includes comprehensive error handling with logging:\n\n```python\nfrom adk_agui_middleware.loggers.logger import setup_logging\n\n# Configure logging\nsetup_logging(level=\"INFO\")\n\n# Error events are automatically generated for:\n# - Agent execution errors\n# - Encoding failures \n# - Tool execution errors\n# - Session management issues\n```\n\n### Tool Integration\n\nHandle tool calls and results seamlessly:\n\n```python\n# Tool calls are automatically extracted from AGUI messages\n# Tool results are processed and converted to the appropriate format\n# No additional configuration required - handled by UserMessageHandler\n```\n\n## API Reference\n\n### Core Classes\n\n- **`SSEService`**: Main service implementation for handling agent interactions\n- **`register_agui_endpoint()`**: Function to register the AGUI endpoint on FastAPI\n- **`ContextConfig`**: Configuration for extracting context from requests\n- **`RunnerConfig`**: Configuration for ADK runner and services\n- **`UserMessageHandler`**: Processes user messages and tool results\n\n### Data Models\n\n- **`SessionParameter`**: Session identification parameters\n- **`BaseEvent`**: Base class for agent events\n- **Various error models**: Structured error handling\n\n## Development\n\n### Code Quality\n\nThe project uses comprehensive linting and formatting:\n\n```bash\n# Format code\nruff format\n\n# Lint code \nruff check\n\n# Type checking\nmypy src/\n```\n\n### Testing\n\n```bash\n# Run tests (configure based on your test setup)\npytest\n\n# With coverage\npytest --cov=adk_agui_middleware\n```\n\n## Configuration\n\n### Environment Variables\n\nConfigure the middleware using environment variables or configuration files:\n\n```bash\n# Logging level\nLOG_LEVEL=INFO\n\n# Service configurations\nUSE_IN_MEMORY_SERVICES=true\nSESSION_SERVICE_URL=redis://localhost:6379\n```\n\n### Runtime Configuration\n\n```python\nfrom adk_agui_middleware.config.log import setup_logging\n\n# Configure logging\nsetup_logging(\n level=\"DEBUG\",\n format=\"json\", # or \"text\"\n output=\"file\" # or \"console\"\n)\n```\n\n## Production Considerations\n\n### Performance\n\n- Use external services (Redis, PostgreSQL) instead of in-memory services\n- Configure appropriate connection pooling\n- Monitor memory usage for long-running sessions\n- Implement proper session cleanup policies\n\n### Security\n\n- Validate all incoming requests\n- Implement proper authentication and authorization\n- Use HTTPS in production\n- Configure CORS appropriately\n- Sanitize user inputs and tool results\n\n### Monitoring\n\n- Monitor agent execution times\n- Track error rates and types\n- Monitor session lifecycle\n- Log security events and access patterns\n\n## Contributing\n\n1. Follow the existing code style and conventions\n2. Add comprehensive docstrings using Google style\n3. Include type annotations for all functions\n4. Write tests for new functionality\n5. Update documentation as needed\n\n### Acknowledgments\n\nThis project is inspired by and builds upon the foundational work of [@contextablemark](https://github.com/contextablemark) and their [AG-UI implementation](https://github.com/contextablemark/ag-ui). We extend our gratitude for their innovative approach to agent-UI integration, which provided essential architectural insights and design patterns that shaped this middleware development.\n\n## License\n\nThis project is licensed under the terms specified in the LICENSE file.\n\n## Support\n\nFor issues and questions:\n1. Check the documentation and examples\n2. Review existing issues in the repository\n3. Create a new issue with detailed information about your use case",
"bugtrack_url": null,
"license": "MIT License\n \n Copyright (c) 2025 DennySORA\n \n Permission is hereby granted, free of charge, to any person obtaining a copy\n of this software and associated documentation files (the \"Software\"), to deal\n in the Software without restriction, including without limitation the rights\n to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\n copies of the Software, and to permit persons to whom the Software is\n furnished to do so, subject to the following conditions:\n \n The above copyright notice and this permission notice shall be included in all\n copies or substantial portions of the Software.\n \n THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\n IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\n FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\n AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\n LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\n OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\n SOFTWARE.",
"summary": "Python middleware bridging Google ADK agents with AGUI protocol via Server-Sent Events for real-time agent interactions",
"version": "0.1.3",
"project_urls": {
"Bug Tracker": "https://github.com/DennySORA/adk-agui-py-middleware/issues",
"Documentation": "https://github.com/DennySORA/adk-agui-py-middleware#readme",
"Homepage": "https://github.com/DennySORA/adk-agui-py-middleware",
"Repository": "https://github.com/DennySORA/adk-agui-py-middleware.git"
},
"split_keywords": [
"agent",
" agui",
" fastapi",
" google-adk",
" middleware",
" sse"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "363c2c25011d78222d45d25ec5a30eddf7f2e08868cde7c35699b8f39729feeb",
"md5": "7b6a065801094d013b1980af3bee7f29",
"sha256": "f9cfbb2b8d591089b7e0983bec257eb5c4c558611da8b6099fc6cb2c179a0bbf"
},
"downloads": -1,
"filename": "adk_agui_py_middleware-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7b6a065801094d013b1980af3bee7f29",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 39255,
"upload_time": "2025-08-13T07:36:20",
"upload_time_iso_8601": "2025-08-13T07:36:20.465073Z",
"url": "https://files.pythonhosted.org/packages/36/3c/2c25011d78222d45d25ec5a30eddf7f2e08868cde7c35699b8f39729feeb/adk_agui_py_middleware-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "dae402b221ba506a0aa548205d036a0940b51d7bb7578bf9622416e576c03184",
"md5": "a51a49a5acc8bb380aaa5295e9bea007",
"sha256": "3bf4d64b33028d091b257950f27ca34f5ac68103d27dd1e2363a8040f691fc80"
},
"downloads": -1,
"filename": "adk_agui_py_middleware-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "a51a49a5acc8bb380aaa5295e9bea007",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 30425,
"upload_time": "2025-08-13T07:36:22",
"upload_time_iso_8601": "2025-08-13T07:36:22.023591Z",
"url": "https://files.pythonhosted.org/packages/da/e4/02b221ba506a0aa548205d036a0940b51d7bb7578bf9622416e576c03184/adk_agui_py_middleware-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-13 07:36:22",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "DennySORA",
"github_project": "adk-agui-py-middleware",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "adk-agui-py-middleware"
}
