agentic-cli 0.4.2

A framework for building domain-specific agentic CLI applications

Agentic CLI

A framework for building domain-specific agentic CLI applications powered by LLM agents.

Overview

Agentic CLI provides the core infrastructure for building interactive CLI applications that leverage LLM agents for complex tasks. It offers:

• Pluggable Orchestration: Choose between Google ADK or LangGraph for agent workflows
• Rich Terminal UI: Thinking boxes, markdown rendering, and streaming responses via thinking-prompt
• Declarative Agents: Define agents with simple configuration objects
• Built-in Tools: Python execution, file operations, knowledge base, web search, web fetch, arXiv search
• Session Persistence: Save and restore conversation sessions
• Type-safe Configuration: Settings management with pydantic-settings

Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                         BaseCLIApp                                  │
│  - Terminal UI (thinking-prompt)                                    │
│  - Command registry (/help, /status, /clear, etc.)                  │
│  - Message history                                                  │
│  - Background initialization (no first-message lag)                 │
│  - Task progress display in thinking box                            │
└─────────────────────────────────────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    BaseWorkflowManager                              │
│  - Agent orchestration                                              │
│  - Event streaming                                                  │
│  - Session management                                               │
├─────────────────────────────┬───────────────────────────────────────┤
│   GoogleADKWorkflowManager  │     LangGraphWorkflowManager          │
│   (Default)                 │     (Optional: langgraph extra)       │
└─────────────────────────────┴───────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────────┐
│                         AgentConfig[]                               │
│  - name, prompt, tools, sub_agents, description                     │
└─────────────────────────────────────────────────────────────────────┘

Requirements

• Python 3.12+
• Google API key (for Gemini models) or Anthropic API key (for Claude models)

Installation

Basic Installation (Google ADK)

pip install agentic-cli

With LangGraph Support

pip install agentic-cli[langgraph]

Development Installation

git clone https://github.com/shoom1/agentic-cli.git
cd agentic-cli
pip install -e ".[dev]"

Using Conda

conda env create -f environment.yml
conda run -n agenticcli pip install -e .

Quick Start

Create a minimal CLI application in just a few lines:

import asyncio
from agentic_cli import BaseCLIApp, BaseSettings
from agentic_cli.cli import AppInfo
from agentic_cli.workflow import AgentConfig

# Define your tools
def greet(name: str) -> dict:
    """Greet a person by name."""
    return {"greeting": f"Hello, {name}!"}

# Configure your agent
AGENTS = [
    AgentConfig(
        name="assistant",
        prompt="You are a helpful assistant. Use the greet tool when asked to greet someone.",
        tools=[greet],
    ),
]

# Create and run your app
if __name__ == "__main__":
    app = BaseCLIApp(
        app_info=AppInfo(name="My App", version="0.1.0"),
        agent_configs=AGENTS,
        settings=BaseSettings(),
    )
    asyncio.run(app.run())

Run with your API key:

export GOOGLE_API_KEY="your-api-key"
python my_app.py

Workflow Managers

GoogleADKWorkflowManager (Default)

Uses Google's Agent Development Kit for orchestration. Best for:

• Simple agent hierarchies
• Google Gemini models with native thinking support
• Quick prototyping

from agentic_cli import GoogleADKWorkflowManager

manager = GoogleADKWorkflowManager(
    agent_configs=AGENTS,
    settings=settings,
    model="gemini-2.0-flash",  # Optional: auto-detected from API keys
)

LangGraphWorkflowManager

Uses LangGraph for orchestration. Best for:

• Cyclical workflows (self-validation, iterative refinement)
• Model-agnostic operation (OpenAI, Anthropic, Google via GenAI)
• State checkpointing and time-travel debugging
• Complex multi-agent coordination

from agentic_cli.workflow.langgraph import LangGraphWorkflowManager

manager = LangGraphWorkflowManager(
    agent_configs=AGENTS,
    settings=settings,
    checkpointer="memory",  # "memory", "postgres", "sqlite", or None
)

Features:

• Explicit provider support: Uses langchain-google-genai for Gemini (not VertexAI)
• Thinking mode: Native support for Claude and Gemini thinking/reasoning
• Retry policies: Automatic retry with exponential backoff
• Event streaming: Real-time workflow events via WorkflowEvent

Requires: pip install agentic-cli[langgraph]

Comparison

Feature	Google ADK	LangGraph
Setup complexity	Simple	Moderate
Cyclical workflows	Limited	Native
Multi-provider	Google only	OpenAI, Anthropic, Google (GenAI)
State persistence	In-memory	Memory, PostgreSQL, or SQLite
Thinking support	Native (Gemini)	Native (Claude & Gemini)
Retry handling	Manual	Built-in with backoff

Auto-selection via Settings

from agentic_cli import create_workflow_manager_from_settings

settings = BaseSettings(orchestrator="langgraph")  # or "adk"
manager = create_workflow_manager_from_settings(agent_configs=AGENTS, settings=settings)

Configuration

BaseSettings

All settings can be configured via environment variables with the AGENTIC_ prefix or in a .env file:

from agentic_cli import BaseSettings

class MySettings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_prefix="MYAPP_",  # Custom prefix
    )
    app_name: str = "my_app"
    workspace_dir: Path = Path.home() / ".my_app"

Key Settings

Setting	Env Variable	Default	Description
google_api_key	GOOGLE_API_KEY	None	Google API key for Gemini
anthropic_api_key	ANTHROPIC_API_KEY	None	Anthropic API key for Claude
default_model	AGENTIC_DEFAULT_MODEL	Auto	Model to use
thinking_effort	AGENTIC_THINKING_EFFORT	"medium"	Thinking level: none, low, medium, high
orchestrator	AGENTIC_ORCHESTRATOR	"adk"	Orchestrator: adk or langgraph
workspace_dir	AGENTIC_WORKSPACE_DIR	~/.agentic	Storage directory
log_level	AGENTIC_LOG_LEVEL	"warning"	Logging level
tavily_api_key	TAVILY_API_KEY	None	Tavily API key for web search
brave_api_key	BRAVE_API_KEY	None	Brave Search API key
search_backend	AGENTIC_SEARCH_BACKEND	Auto	Web search provider (tavily/brave)
webfetch_model	AGENTIC_WEBFETCH_MODEL	Auto	Model for web content summarization

Settings Context

For multi-tenant or isolated contexts:

from agentic_cli import SettingsContext

with SettingsContext(custom_settings):
    # All code here uses custom_settings
    result = my_tool()

Agent Configuration

Agents are defined declaratively using AgentConfig:

from agentic_cli import AgentConfig

# Simple agent
assistant = AgentConfig(
    name="assistant",
    prompt="You are a helpful assistant.",
    tools=[my_tool],
)

# Agent with dynamic prompt
def get_prompt():
    return f"Today is {datetime.now().strftime('%Y-%m-%d')}. Help the user."

dynamic_agent = AgentConfig(
    name="dynamic",
    prompt=get_prompt,  # Callable for dynamic prompts
    tools=[tool_a, tool_b],
)

# Coordinator with sub-agents
coordinator = AgentConfig(
    name="coordinator",
    prompt="Route requests to the appropriate specialist.",
    tools=[],
    sub_agents=["researcher", "analyst"],  # Names of other agents
    description="Routes work to specialists",
)

researcher = AgentConfig(
    name="researcher",
    prompt="Research topics thoroughly.",
    tools=[web_search],
)

analyst = AgentConfig(
    name="analyst",
    prompt="Analyze data and provide insights.",
    tools=[calculate],
)

# Pass all configs to workflow manager
configs = [coordinator, researcher, analyst]

AgentConfig Fields

Field	Type	Description
name	str	Unique identifier
prompt	str | Callable	System instruction
tools	list[Callable]	Available tool functions
sub_agents	list[str]	Names of agents this one can delegate to
description	str	Short description for routing
model	str | None	Model override (defaults to manager's model)

Tools

Creating Tools

Tools are regular Python functions with type hints and docstrings:

def search_database(query: str, limit: int = 10) -> dict:
    """Search the database for matching records.

    Args:
        query: Search query string
        limit: Maximum number of results (default: 10)

    Returns:
        Dict with results and count
    """
    results = db.search(query, limit=limit)
    return {"results": results, "count": len(results)}

Built-in Tools

SafePythonExecutor

Execute Python code in a sandboxed environment:

from agentic_cli.tools import SafePythonExecutor

executor = SafePythonExecutor(default_timeout=30)
result = executor.execute("""
import numpy as np
data = np.array([1, 2, 3, 4, 5])
np.mean(data)
""")
# result = {"success": True, "result": "3.0", "output": "", "error": ""}

Allowed modules: numpy, pandas, scipy, math, json, datetime, collections, itertools, re, random

Web Search

Search the web using pluggable backends (Tavily or Brave):

from agentic_cli.tools import web_search

# Use as agent tool
agent = AgentConfig(
    name="researcher",
    tools=[web_search],
)

# Or call directly
results = web_search("Python async programming", max_results=5)
# Returns: {"results": [{"title": "...", "url": "...", "snippet": "..."}], ...}

Backends auto-select based on available API keys. Set TAVILY_API_KEY or BRAVE_API_KEY.

Web Fetch

Fetch web content and summarize with LLM:

from agentic_cli.tools import web_fetch

result = web_fetch(
    url="https://example.com/article",
    prompt="Extract the main points from this article",
)
# Returns: {"url": "...", "summary": "...", "content_length": ...}

Features: URL validation, robots.txt compliance, SSRF protection, content caching.

ArXiv Search

Search and analyze academic papers:

from agentic_cli.tools import search_arxiv, fetch_arxiv_paper, analyze_arxiv_paper

# Search papers
results = search_arxiv("transformer attention", max_results=10, categories=["cs.CL"])

# Fetch paper details
paper = fetch_arxiv_paper("1706.03762")  # "Attention Is All You Need"

# Analyze with LLM
analysis = await analyze_arxiv_paper("1706.03762", "Summarize the key contributions")

File Operations

Categorized file tools with permission levels:

READ Tools (Safe)

from agentic_cli.tools import read_file, grep, glob, list_dir, diff_compare

# Read file contents
result = read_file("src/main.py", offset=0, limit=100)
# Returns: {"success": True, "content": "...", "size": 1234, "lines_read": 100}

# Search for patterns (ripgrep-like)
result = grep("def.*async", path="src/", file_pattern="*.py", recursive=True)
# Returns: {"success": True, "matches": [...], "file_count": 5}

# Find files by pattern
result = glob("**/*.py", path="src/", include_metadata=True)
# Returns: {"success": True, "files": [...], "count": 42}

# List directory contents
result = list_dir("src/", include_hidden=False)
# Returns: {"success": True, "entries": [...]}

# Compare files or text
result = diff_compare(source1="old.txt", source2="new.txt")
# Returns: {"success": True, "diff": "...", "has_changes": True}

WRITE Tools (Caution)

from agentic_cli.tools import write_file, edit_file

# Write file (creates or overwrites)
result = write_file("output.txt", content="Hello, World!", create_dirs=True)
# Returns: {"success": True, "path": "...", "size": 13, "created": True}

# Edit file (sed-like replacement)
result = edit_file("config.py", old_text="DEBUG = True", new_text="DEBUG = False")
# Returns: {"success": True, "replacements": 1}

Shell Executor

Note: Shell execution is currently disabled by default while security safeguards are being validated.

The shell tool provides layered security with 8 defense layers:

• Input preprocessing (encoding/obfuscation detection)
• Command tokenization and classification
• Path analysis and sandboxing
• Risk assessment with approval workflows
• Audit logging

from agentic_cli.tools import shell_executor, is_shell_enabled

# Check if shell is enabled
if is_shell_enabled():
    result = shell_executor("ls -la", working_dir="/project")
else:
    print("Shell tool disabled pending security validation")

KnowledgeBaseManager

Semantic search over documents:

from agentic_cli.knowledge_base import KnowledgeBaseManager, SourceType

kb = KnowledgeBaseManager(settings=settings)

# Ingest a document
doc = kb.ingest_document(
    content="Machine learning is...",
    title="ML Introduction",
    source_type=SourceType.WEB,
    source_url="https://example.com/ml",
)

# Search
results = kb.search("neural networks", top_k=5)

CLI Commands

Built-in slash commands available in all apps:

Command	Aliases	Description
/help		Show available commands
/status		Show session and workflow status
/clear		Clear the screen
/exit	/quit	Exit the application

Adding Custom Commands

from agentic_cli.cli.commands import Command, CommandCategory

class MyCommand(Command):
    def __init__(self):
        super().__init__(
            name="mycommand",
            description="Do something custom",
            category=CommandCategory.GENERAL,
        )

    async def execute(self, args: str, app: Any) -> None:
        app.session.add_response(f"Executed with args: {args}")

# In your app
class MyApp(BaseCLIApp):
    def get_custom_commands(self) -> list[Command]:
        return [MyCommand()]

Events

WorkflowEvent types for UI integration:

EventType	Description	Metadata
TEXT	Final text response	session_id
THINKING	Model reasoning	session_id
TOOL_CALL	Tool invocation	tool_name, tool_args
TOOL_RESULT	Tool result	tool_name, result, success
CODE_EXECUTION	Code execution result	outcome
ERROR	Error message	recoverable, error_code
USER_INPUT_REQUIRED	Tool needs user input	request_id, prompt
TASK_PROGRESS	Task graph update	current_task_description, progress

Processing Events

async for event in manager.process(message, user_id="user1"):
    if event.type == EventType.TEXT:
        print(event.content)
    elif event.type == EventType.THINKING:
        print(f"[Thinking] {event.content}")
    elif event.type == EventType.TOOL_CALL:
        print(f"Calling: {event.metadata['tool_name']}")
    elif event.type == EventType.TOOL_RESULT:
        print(f"Result: {event.metadata['result']}")

Task Progress Display

When using plan checkboxes or task tools, the CLI thinking box dynamically shows task progress:

Calling: web_search
--- Tasks: 1/3 ---
Research:
  [x] Gather data
  [ ] Analyze results
Writing:
  [ ] Draft report

Status icons (task tools):

• [x] Completed
• [>] In progress
• [ ] Pending
• [-] Cancelled

Examples

See the examples/ directory for complete working examples:

Getting Started

• hello_agent.py - Simple assistant using Google ADK
• hello_langgraph.py - Same assistant using LangGraph orchestration

Feature Demos

• arxiv_demo.py - ArXiv paper search and analysis
• fileops_demo.py - File operation tools (read, write, grep, glob)
• memory_demo.py - Working and long-term memory management
• planning_demo.py - Task graph and planning tools
• shell_demo.py - Shell security pattern detection
• webfetch_demo.py - Web fetching and summarization
• websearch_demo.py - Web search with multiple backends

Full Applications

• research_demo/ - Full-featured research assistant with memory, planning, and file operations

Run examples:

export GOOGLE_API_KEY="your-key"
python examples/hello_agent.py

# Feature demos (no API key needed for some)
python examples/fileops_demo.py
python examples/shell_demo.py

# Or with LangGraph (requires langgraph extra)
pip install agentic-cli[langgraph]
python examples/hello_langgraph.py

# Research demo (full features)
python -m examples.research_demo

Development

Running Tests

# With conda
conda run -n agenticcli python -m pytest tests/ -v

# With pip
pytest tests/ -v

# With coverage
pytest tests/ -v --cov=agentic_cli

Project Structure

agentic-cli/
├── src/agentic_cli/
│   ├── __init__.py           # Package exports
│   ├── config.py             # BaseSettings, SettingsContext
│   ├── constants.py          # Shared constants (truncation, limits)
│   ├── logging.py            # Structlog configuration
│   ├── cli/
│   │   ├── app.py            # BaseCLIApp
│   │   ├── commands.py       # Command, CommandRegistry
│   │   ├── builtin_commands.py
│   │   ├── workflow_controller.py  # Workflow orchestration
│   │   └── message_processor.py    # Event stream processing
│   ├── workflow/
│   │   ├── base_manager.py   # BaseWorkflowManager (abstract)
│   │   ├── events.py         # WorkflowEvent, EventType
│   │   ├── config.py         # AgentConfig
│   │   ├── context.py        # Context variables for tools
│   │   ├── thinking.py       # ThinkingDetector
│   │   ├── adk_manager.py    # GoogleADKWorkflowManager
│   │   └── langgraph/        # LangGraph submodule
│   │       ├── manager.py    # LangGraphWorkflowManager
│   │       ├── state.py      # AgentState, CheckpointData
│   │       └── persistence/  # Checkpointers and stores
│   ├── tools/
│   │   ├── registry.py       # ToolRegistry, ToolCategory, PermissionLevel
│   │   ├── executor.py       # SafePythonExecutor
│   │   ├── file_read.py      # read_file, diff_compare
│   │   ├── file_write.py     # write_file, edit_file
│   │   ├── grep_tool.py      # grep (pattern search)
│   │   ├── glob_tool.py      # glob, list_dir (file discovery)
│   │   ├── search.py         # Web search (Tavily, Brave)
│   │   ├── webfetch_tool.py  # Web content fetching
│   │   ├── memory_tools.py   # MemoryStore, save/search_memory
│   │   ├── planning_tools.py # PlanStore, save/get_plan
│   │   ├── task_tools.py     # TaskStore, save/get_tasks
│   │   ├── hitl_tools.py     # request_approval, create_checkpoint
│   │   ├── standard.py       # ArXiv tools, ask_clarification
│   │   └── shell/            # Shell executor with security
│   │       ├── executor.py   # Main entry point (disabled by default)
│   │       ├── tokenizer.py  # Command parsing
│   │       ├── classifier.py # Risk classification
│   │       ├── sandbox.py    # Execution sandboxing
│   │       └── audit.py      # Security logging
│   ├── knowledge_base/
│   │   ├── manager.py        # KnowledgeBaseManager
│   │   ├── models.py         # Document, SearchResult
│   │   ├── embeddings.py     # EmbeddingService
│   │   ├── vector_store.py   # VectorStore
│   │   └── sources.py        # ArxivSearchSource, SearchSourceRegistry
│   ├── persistence/
│   │   ├── session.py        # SessionPersistence
│   │   ├── artifacts.py      # ArtifactManager
│   │   └── _utils.py         # Atomic write utilities
│   └── hitl/
│       ├── approval.py       # ApprovalManager
│       ├── checkpoints.py    # CheckpointManager
│       └── config.py         # HITLConfig
├── examples/
│   ├── hello_agent.py        # Basic ADK example
│   ├── hello_langgraph.py    # Basic LangGraph example
│   ├── *_demo.py             # Feature demonstration scripts
│   └── research_demo/        # Full-featured example
└── tests/

License

MIT