# Aria Agentic AI Orchestrator Workflow Engine

> Multi-utility AI chatbot combining LangGraph, Gemini 2.5, agentic RAG, intelligent tool routing, persistent threads, MCP integration, and LangSmith observability for production-ready conversational AI.

- **Type:** MCP server
- **Install:** `agentstack add mcp-abhaysingh71-aria-agentic-ai-orchestrator-workflow-engine`
- **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior
- **Seller:** [AbhaySingh71](https://agentstack.voostack.com/s/abhaysingh71)
- **Installs:** 0
- **Category:** [Databases](https://agentstack.voostack.com/c/databases)
- **Latest version:** 0.1.0
- **License:** MIT
- **Upstream author:** [AbhaySingh71](https://github.com/AbhaySingh71)
- **Source:** https://github.com/AbhaySingh71/Aria-Agentic-AI-Orchestrator-Workflow-Engine

## Install

```sh
agentstack add mcp-abhaysingh71-aria-agentic-ai-orchestrator-workflow-engine
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 🤖 Aria - Advanced AI Agent Chatbot

> **A sophisticated multi-utility AI chatbot powered by LangGraph, featuring advanced RAG, intelligent tool calling, MCP integration, and agentic workflow orchestration.**

## Table of Contents
- [Overview](#overview)
- [Architecture](#architecture)
- [Key Features](#key-features)
- [Core Concepts](#core-concepts)
- [Tech Stack](#tech-stack)
- [Project Structure](#project-structure)
- [Installation](#installation)
- [Usage](#usage)
- [Advanced Features](#advanced-features)
- [LangSmith Observability](#langsmith-observability)
- [Contributing](#contributing)

---

## Overview

**Aria** is an intelligent conversational agent that combines the power of:
- **LangGraph**: For orchestrating complex AI workflows
- **LangChain**: For tool management and LLM interactions
- **Google Gemini 2.5**: State-of-the-art language understanding
- **Agentic RAG**: Intelligent document retrieval with reasoning
- **MCP Protocol**: Extensible tool integration via Model Context Protocol
- **SQLite**: Persistent multi-threaded conversation storage

This chatbot goes beyond simple retrieval—it *understands* when to use which tool, combines information from multiple sources, and maintains context across dozens of independent conversations.

---

## Architecture

### System Design

```
┌─────────────────────────────────────────────────────────┐
│                    Streamlit Frontend                    │
│  (Chat UI, PDF Management, Thread Handling)              │
└────────────────┬────────────────────────────────────────┘
                 │
┌────────────────▼────────────────────────────────────────┐
│              LangGraph State Machine                      │
│  ┌──────────────────────────────────────────────────┐   │
│  │ START → chat_node → tools_condition             │   │
│  │           ↑                    ↓                 │   │
│  │           └─── tool_node ←────┘                 │   │
│  └──────────────────────────────────────────────────┘   │
└────────────────┬────────────────────────────────────────┘
                 │
    ┌────────────┼────────────┐
    │            │            │
┌───▼──┐  ┌─────▼──┐  ┌──────▼──────┐
│ LLM  │  │ Tools  │  │ Storage     │
│      │  │        │  │ (SQLite)    │
└──────┘  └────────┘  └─────────────┘
```

### Request Flow

1. **User Input** → Streamlit frontend
2. **Thread Routing** → SQLite retrieves conversation history
3. **LLM Processing** → Gemini determines which tools to use
4. **Tool Execution** → Parallel/sequential tool invocation
5. **Response Assembly** → Stream response back to user
6. **Storage** → Persist conversation to database

---

## Key Features

### 🧠 1. Agentic RAG (Retrieval-Augmented Generation)

**Traditional RAG:**
```
Query → Retrieve → Generate
```

**Aria's Agentic RAG:**
```
Query → Analyze → Decide Retrieval Strategy → Retrieve → Reason → Generate
```

**What makes it "Agentic":**
- **Intelligent Routing**: LLM decides whether RAG is needed or if other tools are better
- **Multi-step Reasoning**: Can combine document insights with web search or calculations
- **Adaptive Chunking**: 1000-character chunks with 200-char overlap for semantic cohesion
- **Semantic Search**: Uses Google's embeddings (models/embedding-001) for similarity matching

**How it works:**
1. User uploads a PDF → Text extraction & splitting
2. Documents are embedded using Google's embedding model
3. FAISS index created for fast similarity search
4. When queried, LLM receives top-4 relevant chunks
5. LLM decides: use RAG info + search + calculate + track expenses?

---

### 🔄 2. Tool Calling System

**7 Integrated Tools:**

| Tool | Purpose | Provider | Type |
|------|---------|----------|------|
| **DuckDuckGo Search** | Real-time web information | DuckDuckGo | External API |
| **Stock Prices** | Financial data retrieval | Alpha Vantage | External API |
| **Calculator** | Math operations | Native | Built-in |
| **RAG Tool** | Document Q&A | FAISS + Embeddings | Internal |
| **Add Expense** | Track spending | MCP Server | Remote Protocol |
| **List Expenses** | View transaction history | MCP Server | Remote Protocol |
| **Summarize** | Generate expense summaries | MCP Server | Remote Protocol |

**Tool Calling Flow:**

```python
LLM Decision:
├─ If ("search" in intent) → DuckDuckGo_Search()
├─ If ("stock" in intent) → Get_Stock_Price()
├─ If ("math" in intent) → Calculator()
├─ If ("document" in intent) → RAG_Tool()
├─ If ("expense" in intent) → MCP_Expense_Tools()
└─ If ("none" in intent) → Direct_Response()
```

**Key Technologies:**
- **LangChain StructuredTool**: Type-safe tool definitions
- **Tool Binding**: LLM knows function signatures, parameters, return types
- **Conditional Edges**: LangGraph routes based on LLM output
- **Async Wrapping**: MCP async tools converted to sync for compatibility

---

### 🔌 3. MCP (Model Context Protocol) Integration

**What is MCP?**
MCP is a standardized protocol for connecting AI models with external tools and data sources. It's like an "API standard" for AI assistants.

**Aria's MCP Setup:**

```
Aria Chatbot ←→ MCP Client ←→ Expense Tracker Server
              (stdio transport)
```

**Local vs Remote:**
- ❌ **Remote Setup Failed**: 401 Unauthorized from remote endpoint
- ✅ **Local Setup Success**: Python subprocess stdio transport

**Architecture:**
```python
# Local MCP Expense Tracker
MCP Service:
├─ add_expense(amount, date, category)
├─ list_expenses(filters)
└─ summarize(period)

Transport: Stdio (stdin/stdout pipes)
Async Handling: StructuredTool wrapper for sync execution
Threading: Separate event loop for async operations
```

**Benefits:**
- **Extensibility**: Add any MCP server without modifying core code
- **Isolation**: Tools run in separate processes
- **Standardization**: Same API whether local or remote

---

### � 6. LangSmith Integration & Observability

**What is LangSmith?**

LangSmith is a unified platform for debugging, testing, and monitoring AI applications. It provides complete visibility into your LLM calls, tool usage, latency, and failures.

**Aria's LangSmith Integration:**

Every chat interaction is automatically traced and logged to LangSmith for real-time monitoring and debugging.

**What Gets Traced:**

```
Chat Turn Trace:
├─ User Input
│  ├─ Message content
│  ├─ Thread ID
│  └─ Message count
├─ Processing
│  ├─ LLM invocation (Gemini 2.5)
│  ├─ Tool decisions & calls
│  └─ Tool execution results
└─ Output
   ├─ AI response
   ├─ Response length
   ├─ Tools used
   └─ Execution timing
```

**Key Metrics Captured:**

| Metric | Purpose | Use Case |
|--------|---------|----------|
| **Response Time** | Track LLM latency | Identify slow responses |
| **Tools Used** | Monitor tool routing | Understand agent behavior |
| **Token Count** | Measure input/output size | Cost analysis |
| **Error Tracking** | Log failures | Debug issues |
| **Thread ID** | Link to conversation | Trace user sessions |
| **Timestamp** | Record exact time | Performance trending |

**LangSmith Dashboard Views:**

1. **Traces Tab**: Complete execution flow for each chat turn
   - Visual hierarchy of LLM calls, tool calls, and responses
   - Input/output for each node
   - Timing information

2. **Metrics Tab**: Aggregated performance data
   - Average response time per conversation
   - Tool usage frequency
   - Error rates and types

3. **Debugging**: Deep-dive into specific traces
   - See exact prompts sent to LLM
   - Tool execution details
   - Error messages and stack traces

**Setup:**

```bash
# Add to .env file
LANGCHAIN_TRACING_V2=true
LANGCHAIN_PROJECT=Advance LangGraph project
LANGCHAIN_API_KEY=
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
```

Get your API key from: https://smith.langchain.com/settings/org/api-keys

**How It Works:**

```python
# Automatic tracing happens in chatbot_stream_with_tracing()
@traceable(name="Chat Turn", tags=["chat", "stream"])
def traced_stream():
    # All code here is automatically traced
    for chunk in chatbot.stream(input_dict, config=config):
        yield chunk
```

Every interaction flows through LangSmith's tracer, creating a complete audit trail of:
- What the user asked
- How the agent reasoned about tools
- Which tools were called and why
- What data was retrieved/computed
- How the final response was generated

**Use Cases:**

1. **Real-time Monitoring**: Watch trace execution as users chat
2. **Performance Analysis**: Identify bottlenecks (slow tools, token usage)
3. **Debugging**: Replay failed traces to understand what went wrong
4. **Cost Tracking**: Monitor LLM token usage and API calls
5. **Quality Assurance**: Audit agent behavior patterns
6. **User Behavior**: Understand which features users interact with

**Example Scenario:**

User: *"Based on my uploaded document, find instances where costs exceeded $10,000"*

**LangSmith Trace Shows:**
- LLM received: user query + system prompt + document chunks (RAG)
- LLM decision: "Need to use RAG tool to search document"
- Tool executed: rag_tool("costs exceeded 10000", thread_id=...)
- Tool returned: 3 matching sections from document
- LLM regenerated: Final response with filtered results
- Trace metadata: response_time=2.3s, tools_used=["rag_tool"], token_count=1240

**Features:**

✅ **Automatic tracing** - No code changes needed, works out of the box
✅ **Tool tracking** - See which tools called and their execution time
✅ **Error capture** - Failures are automatically logged with context
✅ **Session continuity** - Thread IDs link traces to conversations
✅ **No data leakage** - API calls are secure, private to your org

**Dashboard Sample:**

```
Project: Advance LangGraph project
├─ Today
│  ├─ Chat Turn (2:45 PM) - Response: "The document shows..." | Tools: rag_tool | Time: 2.3s
│  ├─ Chat Turn (2:41 PM) - Response: "AAPL is trading at..." | Tools: stock_price | Time: 1.2s
│  ├─ Chat Turn (2:38 PM) - Response: "47 * 32 = 1504" | Tools: calculator | Time: 0.5s
│  └─ Chat Turn (2:35 PM) - Response: "Here's what I found..." | Tools: duckduckgo_search | Time: 3.8s
└─ Last 7 days
   └─ (Historical data...)
```

For detailed configuration and troubleshooting, see [LANGSMITH_FIX.md](LANGSMITH_FIX.md).

---

### �💬 4. Thread Management (Multi-Conversation Support)

**What are Threads?**

Threads enable independent conversation contexts. Each thread has:
- Unique UUID identifier
- Isolated conversation history
- Separate document index (FAISS)
- Dedicated metadata storage

**Per-Thread Storage:**

```
Database (SQLite):
├─ Thread A (UUID: 22258138...)
│  ├─ Message 1: "Hi, analyze my budget"
│  ├─ Message 2: "Create expense tracking"
│  └─ Metadata: {docs_indexed: 1, chunks: 245}
│
├─ Thread B (UUID: 8f9a1d2c...)
│  ├─ Message 1: "Search latest AI news"
│  └─ Message 2: "What about LangGraph?"
│
└─ Thread C (UUID: 7c4e5b9a...)
   ├─ Message 1: "Calculate 500 * 2.5"
   └─ ...
```

**Implementation:**
- **SQLite Checkpointer**: `langgraph_checkpoint_sqlite` persists state
- **Thread ID Config**: Passed to every invocation via `configurable["thread_id"]`
- **Per-Thread Retrievers**: `_THREAD_RETRIEVERS` dict maintains separate FAISS indexes
- **Conversation Isolation**: Messages in Thread A don't affect Thread B

**Thread Operations:**
```python
# Create new thread
thread_id = generate_thread_id()

# Load existing thread
state = chatbot.get_state(config={"configurable": {"thread_id": thread_id}})

# Continue conversation
for msg in chatbot.stream({"messages": [HumanMessage(...)]}, 
                          config={"configurable": {"thread_id": thread_id}}):
    process(msg)
```

---

### 📚 5. Database Architecture

**SQLite with LangGraph Checkpointer**

**Schema** (Auto-managed by langgraph-checkpoint-sqlite):
```
threads:
├─ thread_id (primary key)
├─ checkpoint_data (serialized state)
└─ timestamp

Type structure (internal):
├─ messages: List[Union[HumanMessage, AIMessage, ToolMessage]]
└─ metadata: Dict[str, Any]
```

**Thread-Local Storage:**

```python
# Memory storage (session-based)
_THREAD_RETRIEVERS: Dict[str, Any]  # FAISS indices per thread
_THREAD_METADATA: Dict[str, dict]   # Document info per thread

# Persistent storage (database)
SQLiteSaver checkpointer               # Message history
```

**Persistence Flow:**
```
1. User sends message
2. LLM generates response + tool calls
3. Tools execute
4. State updated: messages + tool outputs
5. SQLiteSaver checkpoints to database
6. Next request loads from checkpoint
```

**Why SQLite?**
- ✅ Simple, no external DB needed
- ✅ Thread-safe (aiosqlite for async)
- ✅ Persistent across restarts
- ✅ Perfect for single-machine deployment
- ✅ Easy migration to PostgreSQL later

---

## Core Concepts

### Concept 1: StateGraph (LangGraph)

**What it is:** A directed acyclic graph (DAG) that defines AI workflow execution.

**Aria's Graph:**
```python
graph = StateGraph(ChatState)
graph.add_node("chat_node", chat_node)           # LLM invocation
graph.add_node("tools", tool_node)               # Tool execution
graph.add_edge(START, "chat_node")               # Entry point
graph.add_conditional_edges(                     # Smart routing
    "chat_node", 
    tools_condition,  # Decides: call tools or finish?
    {
        "continue": "tools",
        END: END
    }
)
graph.add_edge("tools", "chat_node")            # Loop back
```

**Execution Trace:**
```
START
  ↓
chat_node → [LLM decides: need tools?]
  ↓
tools_condition → {
    No tools → END (return response)
    Tools needed → tools
}
  ↓
tools → Execute all tools in parallel
  ↓
chat_node → [LLM processes tool outputs]
  ↓
(loop until no more tools needed)
```

### Concept 2: add_messages Reducer

**Problem:** How do we merge tool outputs back into the message history?

**Solution:** Reducer function
```python
messages: Annotated[list[BaseMessage], add_messages]
```

**What it does:**
```
Old messages:    [User, Assistant, ToolMessage]
New messages:    [ToolMessage, ToolMessage]
Result:          [User, Assistant, ToolMessage, ToolMessage, ...]

(Automatically merges instead of replacing)
```

### Concept 3: Streaming

**Why Stream?**
- ✅ Real-time token-by-token display
- ✅ Better UX (doesn't feel frozen)
- ✅ Faster apparent response time
- ✅ Can cancel long responses

**Aria's Streaming:**
```python
for message_chunk, _ in chatbot.stream(
    {"messages": [HumanMessage(content=user_input)]},
    config=CONFIG,
    stream_mode="messages"
):
    if isinstance(message_chunk, AIMessage):
        yield message_chunk.content  # Stream to Streamlit
```

---

## Tech Stack

### Core Frameworks
- **LangGraph 1.1.6**: AI workflow orchestration
- **LangChain 1.2.15**: LLM abstractions & tools
- **Streamlit 1.48**: Web UI framework
- **SQLite + aiosqlite**: Persistent storage

### AI/ML Components
- **Google Generative AI**: Gemini 2.5 Flash Lite LLM + embedding model
- **FAISS**: Vector similarity search (Facebook AI Similarity Search)
- **Sentence Transformers**: Embeddings pipeline

### Observability & Monitoring
- **LangSmith**: AI application debugging, testing, and monitoring
- **LangSmith Traces**: Real-time execution tracking
- **LangSmith Metrics**: Performance and cost analytics

### Data Processing
- **PyPDF**: PDF text extraction
- **LangChain Text Splitters**: Semantic chunking (1000 chars, 200 overlap)
- **HuggingFace Hub**: Model downloading

### External Integrations
- **DuckDuckGo Search** (ddgs 9.13.0): Web search
- **Alpha Vantage API**: Stock price data
- **MCP Protocol** (fastmcp 3.2.0): Standard tool interface
- **Model Context Protocol** (mcp 1.27.0): Protocol support

### Development Tools
- **Python 3.12.2**: Language version
- **Virtual Environment (venv)**: Dependency isolation

---

## Project Structure

```
Aria/
├── frontend.py                 # Streamlit UI (300+ lines)
│   ├── Sidebar: Thread management, PDF upload
│   ├── Main: Chat interface, features showcase
│   └── CSS: Custom styling
│
├── langgraph_backend.py        # Core orchestration (700+ lines)
│

…

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [AbhaySingh71](https://github.com/AbhaySingh71)
- **Source:** [AbhaySingh71/Aria-Agentic-AI-Orchestrator-Workflow-Engine](https://github.com/AbhaySingh71/Aria-Agentic-AI-Orchestrator-Workflow-Engine)
- **License:** MIT

Install and usage instructions live in the source repository linked above.

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** yes
- **Environment & secrets:** yes
- **Dynamic code execution:** no

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: passed — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/mcp-abhaysingh71-aria-agentic-ai-orchestrator-workflow-engine
- Seller: https://agentstack.voostack.com/s/abhaysingh71
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.