netweb-adk 0.1.6

Netweb Enterprise Agent Engineering Framework

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
• Installation
• Quick Start
• Core Concepts
• Agents
• Tools
• Memory
• Providers
• Streaming
• Tracing and Observability
• Retry and Fault Tolerance
• Examples
• 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

pip install netweb-sdk

Required environment variables are documented at the end of this file.

---

Quick Start

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:

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:

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

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

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

---

Providers

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

• OpenAIProvider
• AnthropicProvider
• GoogleProvider
• OpenSourceProvider (HuggingFace, Ollama, local runtimes)

Example:

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.

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

---

Tracing and Observability

Enable LangSmith-compatible tracing:

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.

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.

Examples

Customer Support (single-agent CLI)

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:

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