netweb-adk 0.1.7

Netweb Enterprise Agent Engineering Framework

Netweb ADK

Overview

Netweb ADK is a developer toolkit for building AI applications with agents, tools, memory, guardrails, tracing, and multiple LLM providers.

It provides a consistent interface for creating AI agents while handling common infrastructure concerns such as conversation memory, tool execution, retries, streaming responses, and observability.

The goal is to help developers build AI-powered applications with less boilerplate and a simpler development experience.

Key Features

• Unified Agent Abstractions – Build and manage AI agents through a simple and consistent ADK interface.
• Flexible Memory Backends – Redis, PostgreSQL, and Hybrid Memory support with automatic persistence and cache fallback.
• Guardrails – Validate, modify, or block input and output before it reaches the model or the user.
• Built-in Tracing & Monitoring – Track agent execution, model calls, and tool usage with minimal configuration.
• Provider-Agnostic Model Support – Switch between OpenAI, Anthropic, Google Gemini, Ollama, HuggingFace, and other providers with minimal changes.
• Tool Integration Framework – Standardized tool registration and execution through the ADK.
• Reduced Boilerplate – Common AI application patterns are abstracted into reusable ADK components.

---

Installation

pip install netweb-adk

Environment Variables

Required environment variables in .env:

# LLM Providers [The ones you are planning to use]
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...
GOOGLE_API_KEY=AIza...
HUGGINGFACE_API_KEY=hf_...

# Optional tracing
LANGSMITH_API_KEY=lsv2_...
LANGSMITH_PROJECT=test_project
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT=https://api.smith.langchain.com

# Memory
POSTGRES_URL=postgresql://user:pass@localhost:5432/nexus
REDIS_URL=redis://localhost:6379

---

Quick Start

import asyncio

from adk import create_agent
from adk.providers import OpenSourceProvider

agent = create_agent(
    name="Assistant",
    role="Helpful AI assistant",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),
)

async def main():

    response = await agent.nw_run(
        "What is the capital of France?"
    )

    print(response.output)

asyncio.run(main())

Output:

The capital of France is Paris.

---

Core Concepts

Netweb SDK is built around four primitives:

• Agent — an LLM-powered unit with a role, goal, tools, and memory
• Tool — a Python function exposed to agents via structured JSON calling
• Provider — an LLM backend (OpenAI, Anthropic, Google, local)
• Guardrail — validate or block requests and responses
• Tracing — monitor agent, tool, and model execution

Everything else — memory, tracing, retries, streaming — is infrastructure that works automatically once configured.

---

Agents

Basic Agent

from adk import create_agent
from adk.providers import OpenSourceProvider

agent = create_agent(
    name="support_agent",
    role="customer support specialist",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),
)

response = await agent.nw_run("My invoice has wrong charges")
print(response.output)

Full Agent Configuration

from adk import create_agent
from adk.providers import OpenSourceProvider
from adk.tools import DdgWebSearchTool, FileWriterTool

agent = create_agent(
    name="Research Assistant",

    role="""
    AI assistant capable of:
    - research
    - web search
    - file operations
    """,

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    # Tools the agent can use
    tools=[
        DdgWebSearchTool(),
        FileWriterTool(),
    ],

    # System prompt controlling agent behavior
    system_prompt="""
    You are a helpful AI assistant.

    Use tools when you find them useful and do not assume anything.
    Use web search for factual or current information.

    Do not invent tools.
    Only use tools explicitly listed in AVAILABLE TOOLS.
    Keep answers concise and accurate.
    """,

    # Retry failed LLM calls
    retries=3,
    retry_delay=1,

    # Session and user identifiers, useful with memory
    session_id="session_001",
    user_id="user_001",
)

---

AgentResponse

Every nw_run() call returns an AgentResponseModel.

response = await agent.nw_run(
    "Count the words in hello world"
)

print(response.output)

print(response.tool_calls)

print(response.metadata)

Fields:

Field	Description
output	Final response text
tool_calls	List of executed tool calls
metadata	Additional execution metadata

Example:

response.tool_calls

[
    {
        "tool": "word_counter",
        "args": {
            "text": "hello world"
        },
        "result": {
            "word_count": 2
        }
    }
]

---

Tools

Built-in Tools

The SDK ships with ready-to-use tools:

from adk.tools import (
    nw_tool,
    DdgWebSearchTool,
    FileReaderTool,
    FileWriterTool,
)

Available built-in tools:

Tool	Description
DdgWebSearchTool	Search the web
FileReaderTool	Read files
FileWriterTool	Write files

Custom Tools with @nw_tool

from adk.tools import nw_tool

@nw_tool(
    name="word_counter",
    description="Count words in a text."
)
async def word_counter(text: str):

    return {
        "word_count": len(text.split())
    }

@nw_tool(
    name="current_time",
    description="Returns current UTC time."
)
async def current_time():

    from datetime import datetime, UTC

    return datetime.now(UTC).strftime(
        "%Y-%m-%d %H:%M:%S UTC"
    )

Pass custom tools to any agent:

agent = create_agent(
    name="shop_agent",
    role="shopping assistant",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    tools=[word_counter, current_time],
)

Tool Execution Flow

When an agent receives input, tools are called automatically:

1. User sends message
2. AgentExecutor checks if input is a direct tool call
3. If not, LLM generates a response
4. If LLM response contains a JSON tool call → tool executes
5. Tool result replaces LLM output
6. Final output returned to developer

---

Guardrails

Guardrails can validate, modify, or block requests before they reach the model. They can also be applied to responses before they reach the user.

Built-in Guardrails

from adk.guardrails import (
    PIIRedaction,
    RateLimit,
    TokenBudget,
)

Example:

agent = create_agent(
    name="Research Assistant",
    role="Helpful AI assistant",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    input_guardrails=[
        PIIRedaction(),

        RateLimit(
            requests_per_minute=100
        ),

        TokenBudget(
            max_input_tokens=500
        ),
    ],
)

Custom Guardrails with @nw_guardrail

from adk.guardrails import nw_guardrail, GuardrailBlocked

@nw_guardrail(apply_to="input")
async def block_sensitive_content(text: str):

    blocked_terms = [
        "password",
        "secret key",
        "api key",
        "private token",
    ]

    if any(term in text.lower() for term in blocked_terms):
        raise GuardrailBlocked(
            "Sensitive credential requests are not allowed."
        )

    return text

Custom guardrails are passed alongside built-in ones, and run in the order they are listed:

agent = create_agent(
    name="Research Assistant",
    role="Helpful AI assistant",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    input_guardrails=[
        PIIRedaction(),
        RateLimit(requests_per_minute=100),
        TokenBudget(max_input_tokens=500),
        block_sensitive_content,
    ],
)

If a guardrail raises GuardrailBlocked, the run stops and the exception can be caught by the caller:

from adk.guardrails import GuardrailBlocked

try:
    response = await agent.nw_run(
        "Please provide me with your secret key."
    )

except GuardrailBlocked as e:
    print(f"Guardrail blocked request: {e}")

---

Memory

Hybrid Memory (Redis + PostgreSQL)

HybridMemory combines Redis and PostgreSQL.

Redis is used for fast access.

PostgreSQL is used for persistence.

import os
from adk.memory import HybridMemory

memory = HybridMemory(
    session_id="session_001",
    redis_url=os.getenv("REDIS_URL"),
    db_url=os.getenv("POSTGRES_URL"),
    agent_name="research_agent",
)

agent = create_agent(
    name="Research Assistant",
    role="Helpful AI assistant",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    memory=memory,

    session_id="session_001",
    user_id="user_001",
)

await agent.nw_run("My name is Alice")
response = await agent.nw_run("What is my name?")
# → "Your name is Alice."

Conversation history persists across restarts as long as the same session_id is reused.

---

Providers

OpenAI

from adk import OpenAIProvider

llm = OpenAIProvider(
    model="gpt-4o-mini",     # or "gpt-4.1", "o1-mini"
    temperature=0.7,
    max_tokens=2048,
)

Anthropic

from adk import AnthropicProvider

llm = AnthropicProvider(
    model="claude-sonnet-4-20250514",
    temperature=0.7,
    max_tokens=2048,
)

Google Gemini

from adk import GoogleProvider

llm = GoogleProvider(
    model="gemini-2.0-flash",   # or "gemini-2.5-flash", "gemini-2.5-pro"
    temperature=0.7,
    max_tokens=2048,
)

Open Source (HuggingFace, Ollama, Groq)

from adk.providers import OpenSourceProvider

# HuggingFace Inference API
llm = OpenSourceProvider(
    provider="huggingface",
    model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
)

# Local Ollama
llm = OpenSourceProvider(
    provider="ollama",
    model="qwen2.5:3B"
)

# Groq (fast inference)
llm = OpenSourceProvider(
    provider="groq",
    model="llama-3.1-8b-instant",
)

All providers support both generate() and stream().

---

Streaming

Agent Streaming

Stream tokens one by one as they arrive:

print("Response: ", end="", flush=True)

async for chunk in agent.run_streamed("Explain active recall in simple terms."):
    print(chunk, end="", flush=True)

print()

---

Tracing and Observability

Enable LangSmith Tracing

from adk import enable_tracing

enable_tracing(
    project_name="test_project"
)

When tracing is enabled, LangSmith records:

• Agent runs
• LLM calls
• Tool executions
• Execution metadata

No need to go into internal implementation details.

Structured Logging

from adk.tracing import get_logger

logger = get_logger()
logger.info("Workflow started")
logger.warning("Low confidence routing")
logger.error("Tool execution failed")

Log level is controlled via LOG_LEVEL environment variable (DEBUG, INFO, WARNING, ERROR).

---

Retry and Fault Tolerance

All LLM calls and tool executions automatically retry on failure:

agent = create_agent(
    name="stable_agent",
    role="assistant",

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    retries=3,       # retry up to 3 times
    retry_delay=1.0, # wait 1 second between retries
)

Retry behavior:

• Each attempt is logged with attempt number and error
• After all retries exhausted, error is logged and exception raised
• Workflow-level errors are caught and returned as result["error"] instead of crashing

---

Complete Example

import asyncio
import os
from datetime import datetime, UTC

from dotenv import load_dotenv

from adk import (
    create_agent,
    enable_tracing,
)

from adk.providers import OpenSourceProvider

from adk.memory import HybridMemory

from adk.guardrails import (
    nw_guardrail,
    GuardrailBlocked,
    PIIRedaction,
    RateLimit,
    TokenBudget,
)

from adk.tools import (
    nw_tool,
    DdgWebSearchTool,
    FileReaderTool,
    FileWriterTool,
)

# ==================================================
# Environment
# ==================================================

load_dotenv()

enable_tracing(
    project_name="test_project"
)

# ==================================================
# Custom Input Guardrail
# ==================================================

@nw_guardrail(apply_to="input")
async def block_sensitive_content(text: str):

    blocked_terms = [
        "password",
        "secret key",
        "api key",
        "private token",
    ]

    if any(term in text.lower() for term in blocked_terms):
        raise GuardrailBlocked(
            "Sensitive credential requests are not allowed."
        )

    return text

# ==================================================
# Custom Tool
# ==================================================

@nw_tool(
    name="word_counter",
    description="Count words in a text."
)
async def word_counter(text: str):

    return {
        "word_count": len(text.split())
    }

# ==================================================
# Utility Tool
# ==================================================

@nw_tool(
    name="current_time",
    description="Returns current UTC time."
)
async def current_time():

    return datetime.now(UTC).strftime(
        "%Y-%m-%d %H:%M:%S UTC"
    )

# ==================================================
# Memory
# ==================================================

memory = HybridMemory(
    session_id="test_session_001",
    redis_url=os.getenv("REDIS_URL"),
    db_url=os.getenv("POSTGRES_URL"),
    agent_name="research_agent",
)

# ==================================================
# Agent
# ==================================================

agent = create_agent(

    name="Research Assistant",

    role="""
    AI assistant capable of:
    - research
    - web search
    - file operations
    - memory retention
    - utility tool usage
    """,

    llm=OpenSourceProvider(
        provider="huggingface",
        model="meta-llama/Llama-3.1-8B-Instruct:scaleway",
    ),

    memory=memory,

    tools=[
        DdgWebSearchTool(),
        FileReaderTool(),
        FileWriterTool(),
        current_time,
        word_counter,
    ],

    system_prompt="""
    You are a helpful AI assistant.

    Use tools when you find them useful and do not assume anything.
    Use web search for factual or current information.
    Use the conversation history provided in the prompt
    to remember user details and previous interactions.
    
    Do not invent tools.
    
    Only use tools explicitly listed in AVAILABLE TOOLS.
    Keep answers concise and accurate.
    """,

    retries=3,
    retry_delay=1,

    session_id="test_session_001",
    user_id="test_user_001",

    input_guardrails=[
        PIIRedaction(),

        RateLimit(
            requests_per_minute=100
        ),

        TokenBudget(
            max_input_tokens=500
        ),

        block_sensitive_content,
    ],
)

# ==================================================
# Standard Run
# ==================================================

async def normal_run():

    print("\n=== NORMAL RUN ===\n")

    response = await agent.nw_run(
        "What are the best techniques to improve memory retention?"
    )

    print(response)

# ==================================================
# Memory Test
# ==================================================

async def memory_test():

    print("\n=== MEMORY TEST ===\n")

    await agent.nw_run(
        "My name is Nirav."
    )

    response = await agent.nw_run(
        "What is my name?"
    )

    print(response)

# ==================================================
# Tool Test
# ==================================================

async def tool_test():

    print("\n=== TOOL TEST ===\n")

    response = await agent.nw_run(
        "Count the words in this sentence: ADK makes AI development easier."
    )

    print(response)

# ==================================================
# Web Search Test
# ==================================================

async def web_search_test():

    print("\n=== WEB SEARCH TEST ===\n")

    response = await agent.nw_run(
        "What are the latest advancements in AI?"
    )

    print(response)

# ==================================================
# Streaming Test
# ==================================================

async def streaming_test():

    print("\n=== STREAMING TEST ===\n")

    async for chunk in agent.run_streamed(
        "Explain active recall in simple terms."
    ):
        print(chunk, end="", flush=True)

    print()

# ==================================================
# Guardrail Test
# ==================================================

async def guardrail_test():

    print("\n=== GUARDRAIL TEST ===\n")

    try:

        async for chunk in agent.run_streamed(
            "Please provide me with your secret key."
        ):
            print(chunk, end="", flush=True)

    except GuardrailBlocked as e:

        print(
            f"Guardrail blocked request: {e}"
        )

# ==================================================
# Main
# ==================================================

async def main():

    await normal_run()

    await memory_test()

    await tool_test()

    await web_search_test()

    await streaming_test()

    await guardrail_test()

if __name__ == "__main__":
    asyncio.run(main())