Metadata-Version: 2.4
Name: netweb-adk
Version: 0.1.7
Summary: Netweb Enterprise Agent Engineering Framework
Author: Netweb AI Team
Keywords: agent,llm,ai,workflow,tooling,enterprise
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: aiofiles==25.1.0
Requires-Dist: asyncpg==0.31.0
Requires-Dist: composio_core==0.7.21
Requires-Dist: ddgs==9.14.2
Requires-Dist: langchain==1.2.17
Requires-Dist: langchain-anthropic==1.4.3
Requires-Dist: langchain-community==0.4.1
Requires-Dist: langchain-google-genai==4.2.2
Requires-Dist: langchain-openai==1.2.1
Requires-Dist: langgraph-checkpoint-postgres==3.1.0
Requires-Dist: langgraph-checkpoint-redis==0.4.1
Requires-Dist: psycopg-binary==3.3.4

<!-- # Netweb SDK

Enterprise-grade AI agent framework focused on building, running, and extending LLM-powered agents, tools, and memory stores — without any workflow or orchestration primitives.

---

## Table of Contents

- [Why Netweb SDK](#why-netweb-sdk)
- [Installation](#installation)
- [Quick Start](#quick-start)
- [Core Concepts](#core-concepts)
- [Agents](#agents)
- [Tools](#tools)
- [Memory](#memory)
- [Providers](#providers)
- [Streaming](#streaming)
- [Tracing and Observability](#tracing-and-observability)
- [Retry and Fault Tolerance](#retry-and-fault-tolerance)
- [Examples](#examples)
- [Environment Variables](#environment-variables)

---

## Why Netweb SDK

Netweb SDK provides a developer-friendly API to build LLM-driven assistants that integrate tools, memory, and provider adapters without forcing a graph/workflow abstraction. It is designed for use-cases where a single agent or a small collection of cooperating agents (coordinated by application code) is sufficient.

Key benefits:

- Structured tool calling and validation
- Pluggable memory backends (in-memory, vector stores, persistent stores)
- Provider-agnostic adapters (OpenAI, Anthropic, Google, local)
- Built-in tracing hooks for observability

---

## Installation

```bash
pip install netweb-sdk
```
Required environment variables are documented at the end of this file.

---

## Quick Start

```python
import asyncio
from dotenv import load_dotenv
from sdk import create_agent, OpenAIProvider

load_dotenv()

agent = create_agent(
    name="assistant",
    llm=OpenAIProvider(model="gpt-4o-mini"),
    memory=True,
)

async def main():
    # use `nw_run` as the convenient blocking helper or `run` within async code
    resp = await agent.nw_run("What is the capital of France?")
    print(resp.output)

asyncio.run(main())
```

---

## Core Concepts

- 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)

Infrastructure features (memory, tracing, retries, streaming) are available without a workflow system.

---

## Agents

Agents are created via `create_agent(...)`. Typical configuration includes the following parameters: `name`, `llm`, `tools`, `memory` (bool or memory instance), `input_schema`, `output_schema`, `prompt_path`, `system_prompt`, `retries`, and `retry_delay`.

Example:

```python
from sdk import create_agent, OpenAIProvider, get_tools

llm = OpenAIProvider(model="gpt-4o-mini")
agent = create_agent(
    name="support_agent",
    llm=llm,
    tools=get_tools(["web_search"]),
    memory=True,
    system_prompt="You are a helpful support assistant.",
    retries=2,
    retry_delay=1.0,
)

async def main():
    resp = await agent.nw_run("My invoice has wrong charges")
    print(resp.output)

asyncio.run(main())
```

AgentResponse contains `output`, `tool_calls`, and `metadata`.

Notes on current behavior
- Memory role: assistant messages stored in memory use the agent's `name` when
    available; if no name is provided the SDK falls back to the literal
    string `assistant`.
- Structured tool calling: when an LLM response includes a JSON object of the
    form {"tool": "<tool_name>", "args": {...}} the SDK will parse that
    JSON and attempt to execute the referenced tool using the exact parameter
    names from the tool schema. The executor removes surrounding Markdown code
    fences before parsing and supports both JSON and Python-literal dicts.
- Streaming + tool output: when streaming is used the LLM chunks are yielded
    as they arrive. After the stream completes the executor inspects the full
    output for a tool call — if a tool was invoked its result is returned as
    an additional chunk (prefixed with a newline).

---

## Tools

Registerable via the `@nw_tool` decorator or available as built-ins. Tools expose a consistent schema, support sync/async execution, and are validated before execution.

Utilities:

- `nw_tool` — decorator to register Python functions as tools
- `get_tools(names: list[str])` — fetch tool instances by name
- `get_tool(name: str)` — fetch a single tool
- `list_tools()` — list available tool metadata

Built-in examples: `web_search`, `file_reader`, `file_writer`, `human_approval`, `escalation`.

Custom tool example:

```python
from sdk import nw_tool

@nw_tool("calculate_discount")
def calculate_discount(price: float, discount_percent: float) -> str:
    final = price - (price * discount_percent / 100)
    return f"Final price after {discount_percent}% discount is ${final:.2f}"
```

---

## Memory

Memory options:

- `ConversationMemory` — in-process conversation history (default when `memory=True`)
- `PersistentMemory` — durable store backed by Postgres/Redis (via checkpointers)

Example (in-memory):

```python
agent = create_agent(..., memory=True)
await agent.nw_run("My name is Alice")
await agent.nw_run("What is my name?")  # → "Your name is Alice."
```

Example (persistent):

```python
from sdk import PersistentMemory
memory = await PersistentMemory.create(db_url=os.getenv("POSTGRES_URL"), thread_id="user_001")
agent = create_agent(..., memory=False)
agent.memory = memory
```

Hybrid memory (Redis + Postgres)

The repository includes a `HybridMemory` implementation that uses Redis as a fast cache layer and Postgres as the source-of-truth. It attempts to read from Redis first and falls back to Postgres. Writes go to Postgres (critical) and Redis (best-effort) so the cache can be rehydrated on reads.

Constructor parameters (important ones):

- `session_id: str` — session/thread identifier used by both layers.
- `redis_url: str` — Redis connection URL for the cache layer.
- `db_url: str` — Postgres connection URL for the durable store.
- `agent_name: str` — optional name used when storing assistant messages (default: `agent`).
- `ttl: int` — Redis TTL in seconds (default: 3600).
- `max_history: int` — number of messages to keep (default: 20).
- `retention_days: int | None` — Postgres retention window (optional).
- `namespace: str` — Redis key namespace (default: `nexus`).
- `table_name: str` — Postgres table used for messages (default: `test_hybrid_memory`).

Example usage:

```python
from sdk import create_agent
from abstractions.memory.hybrid_memory import HybridMemory
import os

hybrid = HybridMemory(
    session_id="user_123",
    redis_url=os.getenv("REDIS_URL"),
    db_url=os.getenv("POSTGRES_URL"),
    agent_name="support_agent",
    ttl=3600,
    max_history=50,
)

agent = create_agent(name="support_agent", llm=llm, memory=hybrid)

# usage remains the same
await agent.nw_run("Remember my name is Alice")
resp = await agent.nw_run("What is my name?")
print(resp.output)
```

Behavior notes:

- On initialization the hybrid memory lazily loads from Redis and, if empty, rehydrates Redis from Postgres.
- `add()` writes to Redis (best-effort) then to Postgres (source-of-truth). If Postgres fails the call raises an error.
- `get()` prefers Redis and falls back to Postgres if Redis is unavailable.
- `clear()`/`clear_async()` clear both layers and reset the internal load flag so the next read will attempt a fresh load.

---

## Providers

Adapters for multiple LLM backends; all support `generate()` and `stream()`.

- `OpenAIProvider`
- `AnthropicProvider`
- `GoogleProvider`
- `OpenSourceProvider` (HuggingFace, Ollama, local runtimes)

Example:

```python
from sdk import OpenAIProvider
llm = OpenAIProvider(model="gpt-4o-mini", temperature=0.7)
```

---

## Streaming

Agents and providers support streaming partial outputs. Use `nw_run_streamed(...)` to iterate chunks as they arrive.

```python
async for chunk in agent.nw_run_streamed("Explain neural networks"):
    print(chunk, end="", flush=True)
```

---



## Tracing and Observability

Enable LangSmith-compatible tracing:

```python
from sdk.tracing import enable_tracing, get_logger
enable_tracing(project_name="my-app")
logger = get_logger()
logger.info("Agent started")
```

Tracing captures LLM call spans, tool executions, and retry attempts.

LangSmith integration
- The SDK includes hooks for LangSmith-compatible tracing. When tracing is
    enabled the executor records chain and LLM spans, and tool executions are
    recorded with metadata (model, prompt/completion token estimates, streaming
    flag, etc.). Set `LANGSMITH_API_KEY` and `LANGSMITH_PROJECT` to enable it.

---

## Retry and Fault Tolerance

Configure per-agent retry policy via `retries` and `retry_delay`. Retries are logged and surfaced in traces.

```python
agent = create_agent(..., retries=3, retry_delay=1.0)
```

Behavioral notes
- The executor will retry transient operations (LLM generate and tool runs)
    according to the agent's `retries` and `retry_delay` settings. Errors are
    logged and surfaced in traces; unrecoverable exceptions are propagated to
    the caller.

---

## Guardrails (Middleware)

This repository includes a small guardrails layer implemented as composable middleware to protect, observe, and modify requests/responses around LLM calls. The middleware lives under `abstractions.guardrails.middleware` and is also exported via the `sdk.guardrails` shim and top-level `sdk` package for convenience.

Available middleware

- `PIIRedactionMiddleware`: redacts PII (emails, phone numbers, credit cards) from the input passed to the LLM. When `redact_pii=True` the middleware temporarily replaces `ctx.raw_input` with a redacted version so downstream handlers receive a sanitized prompt. Optionally set `redact_response=True` to redact response text before it is logged or returned.
- `RateLimitMiddleware`: enforces per-minute and per-day request limits scoped by keys such as `user_id`. Configure `raise_on_exceed` to either raise an exception or return a blocked `AgentResponse`.
- `TokenBudgetMiddleware`: estimates token usage (uses `tiktoken` if available, falls back to a simple character heuristic) and enforces per-request and per-user daily budgets.
- `RetryMiddleware`: retries transient exceptions using fixed/linear/exponential backoff strategies and records attempt counts in `ctx.metadata` (key: `retries.attempt`).

How it works (quick notes)

- Middleware are composed into a `MiddlewareChain` and invoked around a `core_handler` that performs the LLM call. Order matters: `PIIRedactionMiddleware` must appear before the LLM call to ensure the provider receives a sanitized prompt.
- Redaction strategy mutates `ctx.raw_input` for downstream consumers and restores the original value afterward; logs record the redacted input so you can verify what was sent to the provider.
- Rate limit and token budget middleware keep small in-memory state in the demo; in production you should replace these with persistent backends (Redis/Postgres) if you need cross-process limits.

Testing and examples

- Demo server: see `tests/guardrails/test_middleware.py` which builds a demo chain, exposes a FastAPI `/ask` endpoint, and includes a `mock_llm` path useful for retry testing.
- PII redaction check: call the `/ask` endpoint with an input containing an email/phone and inspect the server logs — the log line for the LLM prompt should show `[REDACTED_EMAIL]` / `[REDACTED_PHONE]`.
- Retry testing: the demo `mock_llm` simulates a transient `TimeoutError` on the first attempt for `user_id == "retry_test"`, allowing you to observe retry attempts in the logs and the final successful response.
- Rate-limit & token-budget: adjust the constructor parameters in the demo to lower limits for testing, and choose `raise_on_exceed=False` to have the middleware return a blocked `AgentResponse` rather than raising.

Programmatic assertions

- Direct chain call: call the `CHAIN` instance from the demo to assert retry counts and blocked responses in unit tests.
- Pytest example (suggested): create an async test that runs `CHAIN` with `user_id='retry_test'` and asserts `ctx.metadata['retries.attempt'] >= 1` and that the final `AgentResponse` is not blocked.

Notes

- Real LLM providers require API keys (`HUGGINGFACE_API_KEY`, etc.); the demo uses a mock or local provider for safe testing.
- The middleware implementations in `abstractions.guardrails.middleware` are intentionally simple; extend them for persistent stores, richer redaction rules, or external monitoring as needed.

Guardrails — Implementation & Usage

The SDK provides two related APIs: the low-level middleware (`abstractions.guardrails.middleware`) used by demo servers and tests, and a higher-level unified `NWGuardrail`/builtins layer exported via `abstractions.guardrails`. The built-ins wrap the middleware logic so you can attach guardrails directly to agents.

Built-in guardrails and common constructor parameters:

- `PIIRedaction(apply_to="input"|"output"|"both", redact_pii=True, redact_response=False)` — wraps the PII regex redactor. Use `apply_to="input"` to sanitize prompts before they reach the LLM.
- `RateLimit(apply_to="input", requests_per_minute=None, requests_per_day=None, scope_key="global")` — in-memory sliding-window rate limiter. For cross-process limits implement a persistent backend (Redis/Postgres) in your app and use the middleware directly.
- `TokenBudget(apply_to="both", max_input_tokens=2000, max_output_tokens=None)` — estimates tokens using `tiktoken` when available, falls back to a character heuristic.

You can attach guardrails either by decorator-style registration or programmatically:

Decorator style:

```python
from sdk import create_agent
from abstractions.guardrails.builtins import PIIRedaction, RateLimit, TokenBudget

pii = PIIRedaction(apply_to="input", redact_pii=True)
rl = RateLimit(requests_per_minute=30)
tb = TokenBudget(max_input_tokens=2000)

@pii
@rl
@tb
agent = create_agent(name="support_agent", llm=llm)
```

Programmatic registration (same effect):

```python
agent = create_agent(name="support_agent", llm=llm)
agent.add_guardrail(PIIRedaction(apply_to="input"))
agent.add_guardrail(RateLimit(requests_per_minute=60))
```

Behavior notes & tips:

- `PIIRedaction` mutates the prompt seen by downstream handlers and restores the original before returning — good for preventing PII leakage to providers and logs.
- `RateLimit` and `TokenBudget` use in-memory state by default; for production, use persistent stores or the lower-level middleware hooks to track counts across processes.
- Guardrails raise `GuardrailBlocked` when they intentionally block execution; the agent executor converts this into a blocked `AgentResponse`.


## Examples

### Customer Support (single-agent CLI)

```bash
py -m examples.customer_support.main
```

The example runs a single `support_agent` that uses tools and optional persistent memory. (No workflow or orchestration required.)

### Research + Write (manual chaining)

You can run multiple agents from application code and pass outputs between them:

```python
researcher = create_agent(name="researcher", role="web researcher", llm=llm, tools=get_tools(["web_search"]))
writer = create_agent(name="writer", role="technical writer", llm=llm, tools=get_tools(["file_writer"]))

research_out = await researcher.nw_run("Find benefits of vector search")
write_out = await writer.nw_run(research_out.output)
```

---



## Environment Variables

| Variable | Required | Description |
|---|---|---|
| `OPENAI_API_KEY` | If using OpenAI | OpenAI API key |
| `ANTHROPIC_API_KEY` | If using Anthropic | Anthropic API key |
| `GOOGLE_API_KEY` | If using Google | Google Gemini API key |
| `HUGGINGFACE_API_KEY` | If using HuggingFace | HuggingFace API key |
| `POSTGRES_URL` | If using Postgres memory | `postgresql://user:pass@host:5432/db` |
| `REDIS_URL` | If using Redis memory | `redis://localhost:6379` |
| `LANGSMITH_API_KEY` | If using tracing | LangSmith API key |
| `LANGSMITH_PROJECT` | If using tracing | LangSmith project name |
| `LOG_LEVEL` | No (default: INFO) | `DEBUG`, `INFO`, `WARNING`, `ERROR` |
 -->

<!-- 
# Netweb ADK


# Overview

Netweb ADK is a developer-focused ADK built on top of LangChain that simplifies the creation of AI-powered applications through unified abstractions for agents, memory, tools, tracing, and model providers. It reduces infrastructure and integration complexity by providing reusable components for persistent memory management, observability, tool execution, and provider-agnostic model access, enabling developers to focus on application logic rather than boilerplate code.

# 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.
* **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, 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

```bash
pip install netweb-adk
```


Required environment variables in `.env`:

```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_...

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

# Tracing (optional)
LANGSMITH_API_KEY=lsv2_...
LANGSMITH_PROJECT=my-project
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT=https://api.smith.langchain.com
```

---

## Quick Start

```python
import asyncio
from sdk import create_agent, OpenAIProvider
from dotenv import load_dotenv

load_dotenv()

agent = create_agent(
    name="assistant",
    role="helpful assistant",
    goal="Answer user questions clearly and concisely",
    llm=OpenAIProvider(model="gpt-4o-mini"),
    memory=True,
)

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)

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

---

## Agents

### Basic Agent

```python
from sdk import create_agent, OpenAIProvider

agent = create_agent(
    name="support_agent",
    role="customer support specialist",
    goal="Help customers resolve billing and technical issues",
    llm=OpenAIProvider(model="gpt-4o-mini"),
)

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

### Full Agent Configuration

```python
agent = create_agent(
    name="billing_agent",
    role="billing specialist",
    goal="Handle billing, invoices, and payment queries professionally",
    llm=OpenAIProvider(model="gpt-4o-mini"),

    # Tools the agent can use
    tools=get_tools(["web_search", "file_writer"]),

    # Hard rules the agent must follow
    constraints=[
        "Never share other customer account details",
        "Escalate disputes over $1000 to human review",
        "Never process refunds without verification",
    ],

    # Memory — True for in-RAM, or pass a memory instance
    memory=True,

    # Expected input and output structure
    input_schema={"user_query": str},
    output_schema={"response": str, "confidence": float},

    # Extra behavioral instructions
    extra_instructions="Always end responses with 'Is there anything else I can help you with?'",

    # Load system prompt from a file
    prompt_path="prompts/support/billing.txt",

    # Or provide system prompt directly
    system_prompt="You are a billing specialist at Nexus Corp...",

    # Retry failed LLM calls
    retries=3,
    retry_delay=1.0,
)
```

### AgentResponse

Every `nw_run` call returns an `AgentResponse`:

```python
response = await agent.nw_run("Calculate 20% discount on $250")

response.output      # str — final output text or tool result
response.tool_calls  # list — tools that were called
response.metadata    # dict — {"tool_used": True/False}
```

---

## Tools

### Built-in Tools

The SDK ships with ready-to-use tools:

```python
from sdk import get_tools, get_tool, list_tools

# Get multiple tools by name
tools = get_tools(["web_search", "file_reader", "file_writer"])

# Get a single tool
search = get_tool("web_search")

# List all available tools
all_tools = list_tools()
```

**Available built-in tools:**

| Tool | Description |
|---|---|
| `web_search` | DuckDuckGo web search |
| `file_reader` | Read content from a file |
| `file_writer` | Write content to a file |
| `human_approval` | Pause for human review and approval |
| `escalation` | Escalate cases by severity level |

### Custom Tools with `@nw_tool`

```python
from sdk import nw_tool

@nw_tool
def calculate_discount(price: float, discount_percent: float) -> str:
    """Calculate discounted price."""
    final = price - (price * discount_percent / 100)
    return f"Final price after {discount_percent}% discount is ${final:.2f}"

@nw_tool(description="Get current weather for a city")
async def get_weather(city: str) -> str:
    """Fetch weather data."""
    # your implementation
    return f"Weather in {city}: 22°C, Sunny"

@nw_tool
def calculate_shipping(
    destination: str,
    weight: float,
    express: bool = False
) -> str:
    """Calculate shipping cost."""
    base = 10.0 + (weight * 2.5)
    if express:
        base += 15.0
    return f"Shipping to {destination}: ${base:.2f}"
```

Pass custom tools to any agent:

```python
agent = create_agent(
    name="shop_agent",
    role="shopping assistant",
    goal="Help customers with pricing and shipping",
    llm=OpenAIProvider(),
    tools=[calculate_discount, calculate_shipping, get_weather],
)
```

### 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
```

---

## Memory

### In-RAM Memory (default)

Conversation history stored in Python list. Lost on restart. Zero configuration.

```python
agent = create_agent(
    name="assistant",
    role="assistant",
    goal="Help users",
    llm=OpenAIProvider(),
    memory=True,  # ConversationMemory by default
)

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

### Persistent Memory — Single Agent (Postgres)

Conversation history stored in Postgres. Survives restarts. Readable rows in your database.

```python
import os
from sdk import create_agent, PersistentMemory

memory = await PersistentMemory.create(
    db_url=os.getenv("POSTGRES_URL"),
    thread_id="user_alice_001"  # unique per user/session
)

agent = create_agent(
    name="assistant",
    role="assistant",
    goal="Help users",
    llm=OpenAIProvider(),
    memory=False,
)
agent.memory = memory  # inject persistent memory

# Session 1
await agent.nw_run("My favorite color is blue")

# Session 2 — new agent instance, same thread_id
memory2 = await PersistentMemory.create(
    db_url=os.getenv("POSTGRES_URL"),
    thread_id="user_alice_001"
)
agent2 = create_agent(name="assistant", role="assistant", goal="Help", llm=OpenAIProvider(), memory=False)
agent2.memory = memory2

response = await agent2.nw_run("What is my favorite color?")
# → "Your favorite color is blue."
```

The SDK creates a `nexus_conversation_memory` table automatically:

```sql
SELECT * FROM nexus_conversation_memory WHERE thread_id = 'user_alice_001';
-- id | thread_id       | role      | content                      | created_at
-- 1  | user_alice_001  | user      | My favorite color is blue    | 2026-01-01...
-- 2  | user_alice_001  | assistant | I'll remember that!          | 2026-01-01...
```


---

## Providers

### OpenAI

```python
from sdk import OpenAIProvider

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

### Anthropic

```python
from sdk import AnthropicProvider

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

### Google Gemini

```python
from sdk 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)

```python
from sdk import OpenSourceProvider

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

# Local Ollama
llm = OpenSourceProvider(
    provider="ollama",
    model="llama3.2",
)

# 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:

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

async for chunk in agent.nw_run_streamed("Explain how neural networks work"):
    print(chunk, end="", flush=True)

print()
```

---

## Tracing and Observability

### Enable LangSmith Tracing

```python
from sdk.tracing import enable_tracing

enable_tracing(project_name="my-production-app")
```

After calling `enable_tracing()`, every agent run is automatically traced:

- Full LLM call spans with model name and token counts
- Tool execution spans with inputs and outputs
- Workflow node traces with timing
- Retry attempts and failures

### Structured Logging

```python
from sdk.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:

```python
agent = create_agent(
    name="stable_agent",
    role="assistant",
    goal="Answer questions reliably",
    llm=OpenAIProvider(),
    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

--- -->



# 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

```bash
pip install netweb-adk
```


Environment Variables

Required environment variables in `.env`:

```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

```python
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

```python
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

```python
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`.

```python
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:

```python
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:

```python
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`

```python
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:

```python
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

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

Example:

```python
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`

```python
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:

```python
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:

```python
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.

```python
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

```python
from adk import OpenAIProvider

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

### Anthropic

```python
from adk import AnthropicProvider

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

### Google Gemini

```python
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)

```python
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:

```python
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

```python
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

```python
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:

```python
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

```python
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())
```
