GreenNode AgentBase SDK

A Python SDK for building and deploying AI agents with GreenNode AgentBase runtime.

Overview

GreenNode AgentBase SDK provides a runtime framework for deploying AI agents with support for:

• HTTP endpoint handling for agent invocations
• Health status monitoring and ping endpoints
• Async task tracking
• Request context management
• Streaming responses
• Unified error handling

Installation

pip install greennode-agentbase

Quick Start

from greennode_agentbase import GreenNodeAgentBaseApp

# Create the application
app = GreenNodeAgentBaseApp()

# Define your agent handler
@app.entrypoint
def my_agent(payload: dict):
    """Your agent logic here."""
    query = payload.get("query", "")
    # Process the query with your agent
    return {"response": f"Processed: {query}"}

# Run the server
if __name__ == "__main__":
    app.run(port=8080)

Usage

Basic Agent Handler

from greennode_agentbase import GreenNodeAgentBaseApp

app = GreenNodeAgentBaseApp()

@app.entrypoint
def simple_agent(payload: dict) -> dict:
    """Simple agent that echoes the input."""
    return {"echo": payload}

Handler with Request Context

from greennode_agentbase import GreenNodeAgentBaseApp, RequestContext

app = GreenNodeAgentBaseApp()

@app.entrypoint
def contextual_agent(payload: dict, context: RequestContext) -> dict:
    """Agent that uses request context."""
    session_id = context.session_id
    headers = context.request_headers or {}
    
    return {
        "result": payload,
        "session_id": session_id,
        "authorization": headers.get("Authorization"),
    }

Async Agent Handler

import asyncio
from greennode_agentbase import GreenNodeAgentBaseApp

app = GreenNodeAgentBaseApp()

@app.entrypoint
async def async_agent(payload: dict) -> dict:
    """Async agent handler."""
    # Simulate async work
    await asyncio.sleep(0.1)
    return {"result": "async processing complete"}

Streaming Responses

from greennode_agentbase import GreenNodeAgentBaseApp

app = GreenNodeAgentBaseApp()

@app.entrypoint
def streaming_agent(payload: dict):
    """Agent that streams responses."""
    for i in range(5):
        yield {"chunk": i, "data": f"chunk_{i}"}

Custom Ping Handler

from greennode_agentbase import GreenNodeAgentBaseApp, PingStatus

app = GreenNodeAgentBaseApp()

@app.ping
def custom_health_check() -> PingStatus:
    """Custom health check logic."""
    # Check your system health
    if system_is_busy():
        return PingStatus.HEALTHY_BUSY
    return PingStatus.HEALTHY

Async Task Tracking

from greennode_agentbase import GreenNodeAgentBaseApp

app = GreenNodeAgentBaseApp()

@app.async_task
async def background_processing():
    """Long-running background task."""
    # This task will be tracked for health monitoring
    await process_large_dataset()

Error Handling

All SDK errors inherit from GreenNodeAgentBaseError:

from greennode_agentbase import (
    GreenNodeAgentBaseError,
    GreenNodeValidationError,
    GreenNodeRuntimeError,
    GreenNodeRequestError,
    GreenNodeConfigurationError,
)

try:
    # Your code here
    pass
except GreenNodeValidationError as e:
    print(f"Validation error: {e.message}")
    print(f"Details: {e.details}")
except GreenNodeRuntimeError as e:
    print(f"Runtime error: {e.message}")
except GreenNodeAgentBaseError as e:
    print(f"SDK error: {e.message}")

Command-line interface (CLI)

The greennode CLI provides deploy, memory, and runtime management. Use greennode --help and greennode <group> --help for details.

CLI hierarchy

greennode
├── deploy
│   └── run              Deploy agent (build, push, create runtime). Use -o id to print only the runtime id.
├── memory
│   ├── create           Create a memory
│   ├── list             List memories
│   ├── get              Get a memory by ID
│   └── ...              (other memory subcommands)
└── runtime
    ├── list             List runtimes
    ├── get <id>         Get a runtime by ID
    ├── patch <id>       Update a runtime (image, flavor, env, command, args)
    └── endpoints
        ├── list <runtime-id>     List endpoints for a runtime
        ├── create <runtime-id>   Create an endpoint
        ├── update <runtime-id> <endpoint-id>   Update endpoint (e.g. --version)
        └── delete <runtime-id> <endpoint-id>  Delete an endpoint

Best practices

• Config precedence: Values are resolved in order: environment variables → .greennode.json (for GREENNODE_CLIENT_ID, GREENNODE_CLIENT_SECRET, GREENNODE_ACCESS_CONTROL_NAME, registry vars, etc.). Use .greennode.json for non-sensitive defaults and environment variables for secrets.
• When to use which: Use greennode deploy run for a full deploy (build, push, create runtime). Use greennode runtime list, get, patch and greennode runtime endpoints ... to manage runtimes and endpoints without redeploying.
• Scripting: After greennode deploy run, the runtime id is printed on its own line. Use greennode deploy run -o id to print only the runtime id for scripts.

API Reference

GreenNodeAgentBaseApp

Main application class for deploying agents.

Methods

• entrypoint(func) - Decorator to register the main agent handler
• ping(func) - Decorator to register a custom ping/health handler
• async_task(func) - Decorator to track async tasks
• run(port=8080, host=None) - Start the server
• get_current_ping_status() - Get current health status
• force_ping_status(status) - Force a specific ping status
• add_async_task(name, metadata=None) - Manually track an async task
• complete_async_task(task_id) - Mark an async task as complete

RequestContext

Request context model containing request metadata.

Attributes

• session_id - Optional session identifier
• request_headers - Optional dictionary of request headers
• request - Underlying Starlette request object

GreenNodeAgentBaseContext

Context manager for request-scoped data.

Methods

• set_request_context(request_id, session_id=None) - Set request context
• get_request_id() - Get current request ID
• get_session_id() - Get current session ID
• set_workload_access_token(token) - Set access token
• get_workload_access_token() - Get access token
• set_request_headers(headers) - Set request headers
• get_request_headers() - Get request headers

PingStatus

Enum for health status values:

• HEALTHY - Agent is healthy and ready
• HEALTHY_BUSY - Agent is healthy but processing requests

HTTP Endpoints

POST /invocations

Main endpoint for agent invocations.

Request:

{
  "prompt": "your input here"
}

Response:

{
  "result": "agent output"
}

GET /health

Health check endpoint.

Response:

{
  "status": "Healthy",
  "time_of_last_update": 1234567890
}

Error Handling

All errors in the SDK inherit from GreenNodeAgentBaseError:

• GreenNodeValidationError - Input validation errors
• GreenNodeRuntimeError - Runtime execution errors
• GreenNodeConfigurationError - Configuration errors
• GreenNodeRequestError - HTTP request errors

All errors include:

• message - Error message
• details - Optional error details dictionary
• cause - Optional underlying exception

Development

Setup

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

# Run tests
pytest

# Run linting
ruff check src tests

# Run type checking
mypy src

Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=src --cov-report=html

# Run specific test file
pytest tests/greennode_agentbase/runtime/test_app.py

License

Apache License 2.0 - see LICENSE.txt for details.

Contributing

Contributions are welcome! Please ensure:

• All tests pass
• Code follows the style guidelines (ruff, mypy)
• New features include tests
• Documentation is updated

Support

For issues and questions, please use the project's issue tracker.