PolarSPARC

Agentic CLI From Scratch - Part 2


Bhaskar S 08/15/2026


Overview

In Part 1 of this series, we built the bcode agentic CLI from scratch with support for tools.

In this part, we will extend bcode with support for both conversational memory and skills.


Agentic CLI from Scratch


When we exit and restart bcode, it would start with a clean slate, with no memory of the previous conversation(s).

We will introduce the concept of a session and the ability to save a summary of the conversation on exit. This implies, we will introduce a new slash command /session <uuid>, which will take a previous session <uuid> to restore the conversational context.

Before bcode exits, it compresses the conversation(s) into a summary and saves it in a memory file in the ./memory folder.

Also, we will automatically summarize the conversation(s) once the number of tokens in the context exceeds a threshold value.

The following would be the updated contents of the settings.yaml:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

agent:
  provider: llama.cpp
  model: qwen3.6-a3b
  temperature: 0.9
  top_p: 0.95
  top_k: 64
  base_url: http://192.168.1.25:8000
  tools: True
  context_size_threshold: 1024

tools:
  bash: ask
  edit: allow
  grep: allow
  read: allow
  websearch: ask
  write: allow

The following will be the updated version of the Python script config.py which loads all the configuration settings:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

import logging
import os
import yaml
from pydantic import BaseModel

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

# Default path to the settings file
DEFAULT_SETTINGS_YAML = './settings.yaml'

class LLMConfig(BaseModel):
  """Holds the LLM parameters used to build the chat model"""
  provider: str = 'ollama'
  model: str = 'qwen-3.6'
  temperature: float = 0.0
  top_p: float = 0.5
  top_k: int = 32
  base_url: str = 'http://localhost:11434'
  tools: bool = False
  context_size_threshold: int = 1024

class LLMTools(BaseModel):
  """Holds the LLM tools along with their permissions used to build the chat model"""
  tool_perm: dict[str, str] = {}

def load_config(path: str = DEFAULT_SETTINGS_YAML) -> LLMConfig:
  """Load LLMConfig from a YAML file, falling back to defaults"""
  params = {}
  logger.info(f'Config settings: {path}')
  if os.path.exists(path):
    with open(path, 'r') as settings:
      params = yaml.safe_load(settings) or {}
    logger.info(f'Loaded settings: {params}')
  else:
    logger.info(f'No settings file {path} not found, using defaults')

  return LLMConfig(**params['agent']), LLMTools(tool_perm=params['tools'])

if __name__ == '__main__':
  config, tool_permissions = load_config()
  logger.info(f'config: {config}, tools: {tool_permissions}')

Next, we will need some Python code for the management the memory - to determine the current context size and to help summarize the current conversation(s).

The following is the Python script memory.py which provides the desired functionality:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

import logging
import tiktoken
from langchain_core.messages import HumanMessage

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

_ENCODING = tiktoken.get_encoding('cl100k_base')
SUMMARY_INSTRUCTION = (
  """Summarize the conversation so far in a concise paragraph, preserving important facts, decisions,
  and any open questions. Respond with only the summary text, and nothing else."""
)

def get_context_size(history: list) -> int:
  """Return the token count of the conversation history"""
  text = '\n'.join(str(message.content) for message in history)
  return len(_ENCODING.encode(text))

def summarize_history(agent, history: list) -> str:
  """Ask the agent to summarize the conversation so far as plain text"""
  request = history + [HumanMessage(content=SUMMARY_INSTRUCTION)]
  result = agent.invoke({'messages': request})
  return result['messages'][-1].content

Next, we will need some Python code for the session management - to create a new session id, saving conversational context to memory, and to load the conversational context from a previous session.

The following is the Python script session.py which provides the required functionality:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

import json
import logging
import os
import uuid
from memory import summarize_history

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

DEFAULT_MEMORY_DIR = './memory'

def new_session_id() -> str:
  """Return a fresh session id"""
  return str(uuid.uuid4())

def session_path(session_id: str, memory_dir: str = DEFAULT_MEMORY_DIR) -> str:
  """Return the path to the session memory file for session_id"""
  return os.path.join(memory_dir, f'mem-{session_id}.json')

def save_session(agent, history: list, session_id: str, memory_dir: str = DEFAULT_MEMORY_DIR) -> str:
  """Summarize history and write it to session_id's memory file"""
  summary = summarize_history(agent, history)

  os.makedirs(memory_dir, exist_ok=True)
  path = session_path(session_id, memory_dir)
  with open(path, 'w') as f:
    json.dump({'session_id': session_id, 'summary': summary}, f, indent=2)

  return path

def load_session(session_id: str, memory_dir: str = DEFAULT_MEMORY_DIR) -> str:
  """Return the saved summary for session_id"""
  path = session_path(session_id, memory_dir)
  if not os.path.isfile(path):
    raise FileNotFoundError(f"Session '{session_id}' not found (expected {path})")
  with open(path, 'r') as f:
    data = json.load(f)
  return data['summary']

Finally, we will need to update the agentic CLI to support the conversational memory functionality.

The following is the Python script bcode.py which provides the agentic CLI functionality:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

import logging
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
from langchain.agents import create_agent
from rich.console import Console
from rich.prompt import Prompt
from config import load_config
from llm import build_chat_model
from memory import get_context_size, summarize_history
from session import load_session, new_session_id, save_session
from tools import get_tools

DEFAULT_SYSTEM_MESSAGE = SystemMessage(content='You are a helpful AI task assistant')

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

console = Console()

def initial_history() -> list:
  """Return a clean conversation history containing only the system message"""
  return [DEFAULT_SYSTEM_MESSAGE]

def handle_prompt(agent, history: list, message: str, threshold_size: int):
  """Send the prompt message to the agent and return (response, new_history) the response as tokens
  arrive. Summarize the context when the context size exceeds the threshold"""
  history = history + [HumanMessage(content=message)]

  response = ''
  for chunk, _ in agent.stream({'messages': history}, stream_mode='messages'):
    if isinstance(chunk, AIMessageChunk) and chunk.content:
      console.print(chunk.content, end='')
      response += chunk.content
  console.print()

  new_history = history + [AIMessage(content=response)]
  context_size = get_context_size(new_history)
  if context_size > threshold_size:
    logger.info(f'Context size {context_size} exceeds threshold {threshold_size}, summarizing')
    summary = summarize_history(agent, new_history)
    new_history = [AIMessage(content=summary)]

  return response, new_history

def process_line(line: str, agent, history: list, threshold_size: int):
  """Dispatch the user prompt and return (output_message, new_history, should_continue) tuple.
  If /exit, exit the program"""
  line = line.strip()

  if line == '/exit':
    logger.info("Exiting bcode!")
    return None, history, False

  if line == '/clear':
    logger.info('Context clear command')
    return 'Context Cleared', initial_history(), True

  if line.startswith('/prompt'):
    message = line[len('/prompt'):].strip()
    if not message:
      return 'Usage: /prompt <message>', history, True
    try:
      response, new_history = handle_prompt(agent, history, message, threshold_size)
      return response, new_history, True
    except Exception as exc:
      logger.error(f'Error processing prompt: {exc}')
      return f'Error: {exc}', history, True

  if line.startswith('/session'):
    session_id = line[len('/session'):].strip()
    if not session_id:
      return "Usage: /session <uuid>", history, True
    try:
      summary = load_session(session_id)
    except FileNotFoundError:
      return f"Session '{session_id}' not found", history, True

    new_history = [
      SystemMessage(content=DEFAULT_SYSTEM_MESSAGE),
      SystemMessage(content=f'Conversation Summary:\n{summary}'),
    ]
    return f'Resumed session {session_id}', new_history, True

  return 'Unknown Command', history, True

def run() -> None:
  """Start the agent ReACT loop"""
  logger.info('starting bcode agent CLI')

  config, permissions = load_config()
  logger.info(f'Config: {config}')
  logger.info(f'Permissions: {permissions}')

  llm = build_chat_model(config)

  if config.tools:
    tools = get_tools(permissions)
  else:
    tools = []

  agent = create_agent(llm, tools=tools)
  history = initial_history()

  session_id = new_session_id()
  logger.info(f'Session id: {session_id}')

  console.print('[bold magenta3]-----> bcode v1.2 <-----[/bold magenta3]')
  while True:
    try:
      line = Prompt.ask('[bold dark_orange]bc[/bold dark_orange]', console=console, show_default=False)
    except (EOFError, KeyboardInterrupt):
      console.print()
      break

    if not line.strip():
      continue

    output, history, should_continue = process_line(line, agent, history, config.context_size_threshold)
    if output is not None:
      console.print(output)
    if not should_continue:
      break

  path = save_session(agent, history, session_id)

  logger.info(f'Session {session_id} saved to {path}')
  logger.info('bcode agent stopped')

if __name__ == "__main__":
  run()

Now, to test our agentic CLI bcode, execute the following command in the terminal window:


$ python bcode.py


Once the agentic CLI is ready, the user is presented with the bc: command prompt.

Execute the following /prompt command:


bc: /prompt what does the tool annotation in python langchain for


The agentic CLI will process the prompt and stream the response.

Once again, execute another /prompt command as follows:


bc: /prompt can the same functionality be achieved using other methods

The agentic CLI will process the prompt and stream the response. This time the context threshold is breached and bcode will trigger the summarization logic as shown in the following trimmed output:


Output.1

INFO 2026-08-14 22:09:55,920 - Context size 1573 exceeds threshold 1024, summarizing
INFO 2026-08-14 22:10:18,792 - HTTP Request: POST http://192.168.1.25:8000/chat/completions "HTTP/1.1 200 OK"

Now, execute the /exit command. The following would be the typical output:


Output.2

INFO 2026-08-14 22:11:17,346 - Exiting bcode!
INFO 2026-08-14 22:11:40,921 - HTTP Request: POST http://192.168.1.25:8000/chat/completions "HTTP/1.1 200 OK"
INFO 2026-08-14 22:11:40,922 - Session 383fc1d8-38a7-4178-bdc3-67c6603528c7 saved to ./memory/bcode-383fc1d8-38a7-4178-bdc3-67c6603528c7.json
INFO 2026-08-14 22:11:40,922 - bcode agent stopped

From the Output.2 above, it is evident that the agentic CLI has persisted the session state.

In a terminal window, execute the following command:


ls -l memory


The following would be the typical output:


Output.3

total 4
-rw-rw-r-- 1 polarsparc polarsparc 668 Aug 14 22:11 bcode-383fc1d8-38a7-4178-bdc3-67c6603528c7.json

Now, let us iterate to the final version of bcode with support for skills !

Think of a skill as a packaged set of instructions, defined in a markdown file (skill.md), which tells bcode how to do a specific task. The contents of skill.md get loaded into the context of bcode for processing.

Each skill will have a name and will be created as a folder with that name in the ./skills folder. For example, a skill named summarize will have an associated skill.md in the folder ./skills/summarize.

The following will be the contents of the skill ./skills/summarize/skill.md:


---
name: summarize
description: Summarize the conversation so far into concise bullet points covering key facts, decisions, and open questions.
---

Summarize the conversation so far in three concise bullet points, highlighting the key facts, decisions, and any open questions.

We will need some Python code to load the instructions for a specific skill. This is where the following Python script skills.py comes into the picture:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

import frontmatter
import logging
import os

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

DEFAULT_SKILLS_DIR = './skills'
SKILL_FILENAME = 'skill.md'

class SkillError(Exception):
  """Raised when a skill.md file is missing or malformed."""

def _parse_skill_md(path: str, folder_name: str) -> tuple:
  """Parse a skill.md file into three parts (name, description, body)"""
  md = frontmatter.load(path)

  name = md.get('name')
  if not name:
    raise SkillError(f"{path}: Frontmatter missing mandatory 'name' field")
  if md.get('name') != folder_name:
    raise SkillError(f"{path}: Frontmatter 'name' ({name!r}) must match skill folder ({folder_name!r})")

  description = md.get('description')
  if not description:
    raise SkillError(f"{path}: Frontmatter missing mandatory 'description' field")

  return name, description, md.content.strip()

def load_skill(name: str, skills_dir: str = DEFAULT_SKILLS_DIR) -> str:
  """Return the body of skills//skill.md (frontmatter stripped)"""
  skill_path = os.path.join(skills_dir, name, SKILL_FILENAME)
  if not os.path.isfile(skill_path):
    raise FileNotFoundError(f"Skill '{name}' not found (expected {skill_path})")
  _, _, body = _parse_skill_md(skill_path, name)
  return body

Finally, we will need to update Python code for the agentic CLI to use the tools. The following is the final version for the Python script bcode.py:


#
# @Author: Bhaskar S
# @Blog:   https://polarsparc.github.io
# @Date:   14 Aug 2026
# @Ver:    1.2
#

import logging
from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
from langchain.agents import create_agent
from rich.console import Console
from rich.prompt import Prompt
from config import load_config
from llm import build_chat_model
from memory import get_context_size, summarize_history
from session import load_session, new_session_id, save_session
from skills import load_skill, SkillError
from tools import get_tools

DEFAULT_SYSTEM_MESSAGE = SystemMessage(content='You are a helpful AI task assistant')

logging.basicConfig(format='%(levelname)s %(asctime)s - %(message)s', level=logging.INFO)
logger = logging.getLogger(__name__)

console = Console()

def initial_history() -> list:
  """Return a clean conversation history containing only the system message"""
  return [DEFAULT_SYSTEM_MESSAGE]

def handle_prompt(agent, history: list, message: str, threshold_size: int):
  """Send the prompt message to the agent and return (response, new_history) the response as tokens
  arrive. Summarize the context when the context size exceeds the threshold"""
  history = history + [HumanMessage(content=message)]

  response = ''
  for chunk, _ in agent.stream({'messages': history}, stream_mode='messages'):
    if isinstance(chunk, AIMessageChunk) and chunk.content:
      console.print(chunk.content, end='')
      response += chunk.content
  console.print()

  new_history = history + [AIMessage(content=response)]
  context_size = get_context_size(new_history)
  if context_size > threshold_size:
    logger.info(f'Context size {context_size} exceeds threshold {threshold_size}, summarizing')
    summary = summarize_history(agent, new_history)
    new_history = [AIMessage(content=summary)]

  return response, new_history

def process_line(line: str, agent, history: list, threshold_size: int):
  """Dispatch the user prompt and return (output_message, new_history, should_continue) tuple.
  If /exit, exit the program"""
  line = line.strip()

  if line == '/exit':
    logger.info("Exiting bcode!")
    return None, history, False

  if line == '/clear':
    logger.info('Context clear command')
    return 'Context Cleared', initial_history(), True

  if line.startswith('/prompt'):
    message = line[len('/prompt'):].strip()
    if not message:
      return 'Usage: /prompt <message>', history, True
    try:
      response, new_history = handle_prompt(agent, history, message, threshold_size)
      # We will return a None for response as the response has already been displayed
      # on the console due to streaming chunks
      return None, new_history, True
    except Exception as exc:
      logger.error(f'Error processing prompt: {exc}')
      return f'Error: {exc}', history, True

  if line.startswith('/skill'):
    name = line[len('/skill'):].strip()
    if not name:
      return 'Usage: /skill <name>', history, True
    try:
      message_body = load_skill(name)
      response, new_history = handle_prompt(agent, history, message_body, threshold_size)
      # We will return a None for response as the response has already been displayed
      # on the console due to streaming chunks
      return None, new_history, True
    except FileNotFoundError:
      error_msg = f"Unknown skill '{name}'"
      logger.error(error_msg)
      return error_msg, history, True
    except SkillError as exc:
      error_msg = f"Error processing skill '{name}': {exc}"
      logger.error(error_msg)
      return error_msg, history, True

  if line.startswith('/session'):
    session_id = line[len('/session'):].strip()
    if not session_id:
      return "Usage: /session <uuid>", history, True
    try:
      summary = load_session(session_id)
    except FileNotFoundError:
      return f"Session '{session_id}' not found", history, True

    new_history = [
      SystemMessage(content=DEFAULT_SYSTEM_MESSAGE),
      SystemMessage(content=f'Conversation Summary:\n{summary}'),
    ]
    return f'Resumed session {session_id}', new_history, True

  return 'Unknown Command', history, True

def run() -> None:
  """Start the agent ReACT loop"""
  logger.info('starting bcode agent CLI')

  config, permissions = load_config()
  logger.info(f'Config: {config}')
  logger.info(f'Permissions: {permissions}')

  llm = build_chat_model(config)

  if config.tools:
    tools = get_tools(permissions)
  else:
    tools = []

  agent = create_agent(llm, tools=tools)
  history = initial_history()

  session_id = new_session_id()
  logger.info(f'Session id: {session_id}')

  console.print('[bold magenta3]-----> bcode v1.3 <-----[/bold magenta3]')
  while True:
    try:
      line = Prompt.ask('[bold dark_orange]bc[/bold dark_orange]', console=console, show_default=False)
    except (EOFError, KeyboardInterrupt):
      console.print()
      break

    if not line.strip():
      continue

    output, history, should_continue = process_line(line, agent, history, config.context_size_threshold)
    if output is not None:
      console.print(output)
    if not should_continue:
      break

  path = save_session(agent, history, session_id)

  logger.info(f'Session {session_id} saved to {path}')
  logger.info('bcode agent stopped')

if __name__ == "__main__":
  run()

To test our agentic CLI bcode, execute the following command in the terminal window:


$ python bcode.py


Once the agentic CLI is ready, the user is presented with the bc: command prompt.

Execute the following /prompt command:


bc: /prompt what does the tool annotation in python langchain for


The agentic CLI will process the prompt and stream the response.

Now, execute the following /skill command:


bc: /skill summarize


The following should be the typical trimmed output:


Output.4

INFO 2026-08-14 23:33:18,563 - HTTP Request: POST http://192.168.1.25:8000/chat/completions "HTTP/1.1 200 OK"
- **Key Fact:** The `@tool` decorator in LangChain automatically converts Python functions into LLM-ready tools by generating Pydantic argument schemas, enforcing type validation, and using docstrings as natural language descriptions for the model.
- **Technical Note/Decision:** Modern LangChain (`>=0.2`) explicitly recommends importing from `langchain_core.tools` rather than the deprecated `langchain.tools` namespace to ensure proper async support, type safety, and long-term compatibility.
- **Open Question/Next Step:** The conversation is currently conceptual; practical implementation details (e.g., wiring `@tool` into a specific agent framework like `create_tool_calling_agent` or handling tool error/retry logic) remain open pending your specific use case.

### want to add support for agent and subagents in the bcode cli. do not make changes just show me the approach

Executing the following /exit command will exit the agentic CLI bcode !

Will leave it to the explorer(s) to iterate the next version of bcode to support agent(s) !

An agent is very similar to a skill, defined in a markdown file (agent.md), but with its own conversational context (independent of the main context).

With this, we conclude Part 2 of the series on building an agentic CLI from scratch !!!


References

Agentic CLI From Scratch - Part 1

GitHub - BCode Repository



© PolarSPARC