Gonzalo Ayuso – Web Architect

A Pokédex in the terminal, but agentic, with LangChain

Gonzalo Ayuso — Mon, 29 Jun 2026 12:13:18 +0000

This project is a spin-off of a small that we started during the Katayuno on June 20, 2024. Katayuno is a Saturday morning programming kata where the conversation, debate and retrospective normally matter more than finishing the exercise. That time, helped by AI, almost every team actually finished.

My version used React, TypeScript and Vite, with PokeAPI directly from the browser React Pokédex. It had the usual things: list, details, comparison, shiny sprites and even a small battle mode.

But that application was deterministic. Click here, fetch this, render that.

This time I wanted something slightly different. I wanted a Pokémon professor in my terminal. Not a chatbot that hallucinates Pokémon facts. A small agent that uses PokeAPI as the source of truth and an LLM only as the reasoning layer.

The LLM is not the database. The LLM is the reasoning layer.

I don’t want the model to remember Pikachu’s Speed. I want the model to ask the tool.

The idea

The project is a Python 3.13 CLI built with Click, Rich, httpx, Pydantic and LangChain. AWS Bedrock is the default model provider, but it lives behind one small factory, so changing the provider does not require changing the PokeAPI code.

The flow is deliberately boring:

The agent cannot fetch arbitrary URLs. It only receives five tools:

get_pokemon
get_type
compare_pokemon
get_evolution_chain
get_type_matchup

This is not magic. The tools are normal Python functions returning small, normalized dictionaries. LangChain decides when to call them and the model reasons with their output.

The CLI

The deterministic commands work without AWS credentials:

uv run python -m cli pokemon pikachu
uv run python -m cli search charizrad
uv run python -m cli compare charizard blastoise

The commands involving reasoning can use Bedrock:

uv run python -m cli compare charizard blastoise --explain
uv run python -m cli battle charizard venusaur
uv run python -m cli ask "Which Pokémon is faster, Gengar or Alakazam?"

The boring deterministic part

Getting a Pokémon by name does not need an LLM. This is one of those examples where using an LLM for everything would be a mistake.

The PokeAPI adapter turns the large API response into a small Pydantic model:

			
class PokemonSummary(BaseModel):
    id: int
    name: str
    types: list[str]
    height: int = Field(description="Height in decimetres, as returned by PokeAPI")
    weight: int = Field(description="Weight in hectograms, as returned by PokeAPI")
    base_experience: int | None
    stats: list[PokemonStat]
    abilities: list[str]
    def stat(self, name: str) -> int:
        return next((stat.value for stat in self.stats if stat.name == name), 0)
    @property
    def total_stats(self) -> int:
        return sum(stat.value for stat in self.stats)

		

The pokemon command fetches the data and Rich renders it. The compare command fetches two Pokémon and compares their base stats with plain Python.

The deterministic part is boring. And that is good.

It is easy to test and easy to understand when something goes wrong.

Typos are deterministic too. A failed lookup downloads the compact species-name index from PokeAPI and keeps it in memory for the current process. Python’s SequenceMatcher then ranks names locally:

uv run python -m cli search charizrad

charizard 89% similarity

Normal commands use the same matcher when PokeAPI returns a 404:

╭──────────────────────────── Pokémon not found ─────────────────────────────╮
│ I couldn't find 'charizrad'.                                               │
│                                                                            │
│ Best match: Charizard  (89% similarity)                                    │
╰────────────────────────────────────────────────────────────────────────────╯
Press Enter to use Charizard, type another name, or q to cancel:

Pressing Enter accepts the suggestion. Typing another name retries the lookup, and q cancels. I deliberately do not ask the LLM and I do not silently replace the name. In non-interactive scripts the command keeps returning a normal error instead of waiting forever for input.

The agentic part

The agentic part starts when the question is no longer a direct API call.

For example:

Which Pokémon is faster, Gengar or Alakazam?
Can Pikachu beat Squirtle?
What are Dragonite weaknesses?
Tell me the evolution chain of Eevee

Here LangChain’s create_agent receives the Bedrock chat model, the controlled tools and a system prompt:

			
agent = create_agent(
    model=create_chat_model(settings),
    tools=build_tools(client),
    system_prompt=SYSTEM_PROMPT,
)
result = agent.invoke(
    {"messages": [{"role": "user", "content": question}]}
)

		

The important part is not create_agent. The important part is the boundary. Facts come from tools. The model decides which facts it needs and explains the result.

The prompt says it explicitly:

You must never invent Pokémon data. Use the available tools to retrieve facts
from PokeAPI before answering factual questions.

The LLM is not the database. The LLM is the reasoning layer.

A prompt is not a security boundary, of course. That is why the agent only gets small, explicit tools and never gets a generic HTTP client.

Tools

The tools are created around the PokeAPI client. This makes them small and also makes tests simple because I can inject an httpx.MockTransport.

			
@tool
def get_type_matchup(
    attacker_type: str,
    defender_types: list[str],
) -> dict[str, Any]:
    """Calculate the damage multiplier for one attacking type against defender types."""
    return client.get_type_matchup(
        attacker_type,
        defender_types,
    ).model_dump()

		

I don’t return the complete PokeAPI JSON. Agents work better when tools return the information needed for the task instead of a small novel containing every field an API has accumulated over the years.

Structured output

The battle command is intentionally limited. It is not a competitive Pokémon simulator. It considers the Pokémon types, type multipliers, base Speed, offensive stats and defensive stats. It does not consider moves, levels, abilities, held items, natures, weather or battle format.

The local heuristic first creates a valid prediction:

			
class BattlePrediction(BaseModel):
    winner: str
    confidence: float = Field(ge=0, le=1)
    reasons: list[str]
    caveats: list[str]
    recommended_attack_types: list[str]

		

In Bedrock mode I pass that prediction and the normalized PokeAPI facts to a second LangChain agent using response_format=BattlePrediction. The result is validated by Pydantic instead of parsing an optimistic blob of JSON from a string.

			
agent = create_agent(
    model=create_chat_model(settings),
    tools=[],
    system_prompt=BATTLE_PROMPT,
    response_format=ToolStrategy(BattlePrediction),
)

		

The model can improve the explanation, but it does not get permission to invent a Flamethrower, an item or a hidden ability.

I use LangChain’s ToolStrategy explicitly here. Bedrock’s native structured output currently rejects some numeric JSON Schema constraints generated by Pydantic, such as the minimum and maximum for confidence. Tool calling still returns a validated BattlePrediction without depending on that provider limitation.

Rich output

Click handles the command-line interface and Rich handles tables, panels, colours and stat bars.

Rich is not needed, but terminals should still look decent.

The visual layer is also separate from the data layer. render.py receives Pydantic models. It does not know how PokeAPI works and it does not call the LLM.

When not to use the LLM

I think this is the useful part of the experiment.

There is no model call in:

pokemon
compare without --explain
typo suggestions and Pokémon name search
type multiplier calculation
the first battle prediction
tests

An LLM is useful when the user asks an open question and the application needs to choose tools, combine facts and explain a conclusion. It is not useful for adding six integers or reading Pikachu’s height from JSON.

Using less AI here makes the agentic part easier to see.

Running the project

I normally use Poetry. For this small project I wanted to try uv, so it owns Python installation, dependency resolution, command execution and the lock file. I am not starting a package-manager religion here. It’s just a test.

git clone https://github.com/gonzalo123/pokemon_cli.git
cd pokemon_cli

uv python install 3.13
uv sync --extra dev

That is enough for the deterministic commands.

For AWS Bedrock:

uv sync --extra dev --extra bedrock
cp .env.example .env

Then configure the environment:

AWS_PROFILE=sandbox
AWS_REGION=eu-west-1
BEDROCK_MODEL_ID=global.anthropic.claude-sonnet-4-6

No AWS key is stored in the repository. The AWS SDK uses the selected profile or its normal credential chain.

Now the examples:

uv run python -m cli pokemon pikachu
uv run python -m cli compare charizard blastoise
uv run python -m cli battle charizard venusaur
uv run python -m cli ask \
  "Which Pokémon is faster, Gengar or Alakazam?"

There is also an installed command:

uv run pokemon-professor pokemon pikachu

Things I liked

The separation is small but useful. PokeAPI owns the facts, Pydantic owns the shape, Python owns deterministic calculations, Rich owns presentation and the LLM owns a narrow reasoning task.

Things that still feel awkward

The battle result is only a heuristic. A real battle model needs moves, abilities, levels, items, natures, generation rules, and probably much more. Adding all that while pretending the result is still simple would be dishonest.

I’m not a Pokémon expert. I have only been playing with the Pokémon API because of the Katayuno. This is just an excuse to use AI everywhere, just like we developers seem to be doing these days. Please don’t judge me too harshly.

Final thoughts

Agentic does not mean replacing every function with an LLM call. For me it means giving the model a small set of reliable capabilities and letting it use them when a deterministic route is no longer enough.

The Pokémon facts do not belong in the prompt and they do not belong in the model’s memory. They belong in PokeAPI.

The LLM is not the database. The LLM is the reasoning layer.

And that’s all. Full source code is available in my GitHub account.

What Homer Would Reply to Your Email: A Classical Quote Recommender with Bedrock, RAG, and Strands Agents

Gonzalo Ayuso — Mon, 08 Jun 2026 12:18:45 +0000

What if every email you write could carry the rhetorical weight of Homer? Not as a gimmick, as a real tool that understands the tone, intent, and emotion of your message and finds the classical passage that fits.

That’s the idea behind this PoC: a system that takes a short text (an email, a Slack message, a reply to a tricky thread) and recommends the most fitting quote from classical literature. Right now the corpus is the Iliad and the Odyssey, nearly 5,000 passages of Homer, indexed and searchable by meaning, not just by keywords.

The interesting part isn’t the concept. It’s how the pieces fit together.

The Architecture

The system runs as a FastAPI service with a five-stage pipeline:

The key idea: the two ends of the pipeline (understanding your message and choosing the final quote) use an LLM, but the middle part, finding and ranking candidates, is pure code, no AI involved. That means the core search is predictable and inspectable. Bedrock handles the parts that need language understanding; everything else stays local.

Query Enrichment: Not Just What You Said, But What You Meant

When you search a typical RAG system, you take the user’s text and look for similar documents. Here I do something different: before searching, the system enriches the query with everything it learned during the rhetorical analysis.

			
class HybridRetriever:
    def _build_query(self, text: str, analysis: InputAnalysis) -> str:
        parts = [
            text,
            analysis.summary,
            analysis.main_theme,
            " ".join(analysis.secondary_themes),
            analysis.tone,
            analysis.intent,
            analysis.dominant_emotion,
            analysis.recommended_quote_type,
        ]
        return " ".join(part for part in parts if part).strip()

		

Say the input is “Thanks for your feedback. I think we can find a middle ground”. Instead of just searching for those words, the system searches for negotiation, conciliatory, guarded optimism, diplomatic and bridge-building, all at once. The search doesn’t just look for passages about feedback. It looks for passages that feel the same way as the message.

Zero-Dependency Vector Store

I deliberately avoided FAISS, Chroma, Pinecone, or any external vector database. The entire search index is a single NumPy matrix:

			
class NumpyVectorStore:
    def search(self, query_vector: np.ndarray, top_k: int) -> list[tuple[str, float]]:
        query = np.asarray(query_vector, dtype=np.float32)
        scores = self.vectors @ query
        indices = np.argsort(scores)[::-1][:top_k]
        return [(self.ids[index], float(scores[index])) for index in indices]

		

One line does the work: scores = self.vectors @ query. This is a matrix multiplication that compares the query against every passage in the corpus at once. Because the embedding model produces normalized vectors, this simple operation gives you cosine similarity, a standard way to measure how close two texts are in meaning.

The full index is a matrix of 4,841 rows (one per passage) and 384 columns (one per dimension of the embedding). It loads in milliseconds and fits in memory easily. For a corpus of this size, a full-blown vector database would be overkill.

Why all-MiniLM-L6-v2

The model that turns text into numbers (embeddings) is all-MiniLM-L6-v2 from Sentence Transformers. It takes any piece of text and produces a list of 384 numbers that represent its meaning. Texts that say similar things end up with similar numbers, even if they use completely different words.

It’s a small model, only 22 million parameters and 6 layers, but it was trained on over a billion pairs of sentences, so it’s surprisingly good at capturing semantic similarity. It loads in under a second on a regular CPU and processes the entire Homer corpus in a few seconds.

There are bigger models (like all-mpnet-base-v2 with 110M parameters) that would be slightly more precise, but for this project the bottleneck isn’t the embedding quality, it’s how well the query enrichment captures the intent of the message. The small model is more than enough.

The Reranker: Five Signals, Calibrated Weights

After the search phase finds candidate passages using both meaning and keywords, a rule-based reranker scores each one across five signals:

			
candidate.rerank_score = (
    0.45 * candidate.hybrid_score
    + 0.20 * candidate.thematic_fit
    + 0.15 * candidate.tonal_fit
    + 0.10 * candidate.clarity_fit
    + 0.10 * candidate.rhetorical_fit
)

		

Each signal measures something different: how well the topic matches, whether the tone is right, whether the passage would work rhetorically. The most product-shaped one is clarity_fit. It now favors excerpts that are short, sentence-bounded, and easy to paste into an email without further editing.

That matters because pure semantic similarity misses a very obvious real-world constraint: a 200-word passage from Homer might be relevant, but it’s still useless if what you need is a quotable closing line.

Strands Agents with Structured Output

The analyzer and selector use Strands Agents, an open-source SDK for building AI agents. The key feature I rely on is structured output: instead of asking the LLM to return free text and then parsing it, the SDK forces the model to fill in a typed Python object directly.

			
agent = Agent(
    model=self.model,
    system_prompt=(
        "You analyze short emails or messages for a rhetorical quote recommender. "
        "First infer the input language from the text itself. "
        "Return the detected language plus concise, grounded rhetorical analysis."
    ),
)
result = agent(prompt, structured_output_model=AnalyzerResult)

		

The AnalyzerResult is a Pydantic model with fields like main_theme, tone, intent, dominant_emotion, and recommended_quote_type. The model doesn’t write prose, it fills in a form. This eliminates a whole category of bugs related to parsing LLM output.

The selector agent works the same way: it receives the ranked candidates as JSON and returns exactly three choices, each with a why_it_fits explanation written in the language of the original message.

Compact Quotes, Not Paragraphs

One thing became obvious very quickly: finding a semantically relevant passage is not the same as finding a quotable one. For emails and short messages, a 70-word block of Homer is basically unusable.

So before the selector sees the candidates, the reranker extracts a compact quote window from each passage, usually one or two sentences, under 32 words. The quote still comes verbatim from the indexed corpus, but the system stops treating the full chunk as the thing you’d actually paste into an email.

That small layer makes a big difference. It biases the output toward something you can actually use without turning your message into a wall of text.

Representative Examples

Here are two representative examples of the kind of output the system is designed to produce.

Example 1

Input

Thanks for your feedback. I do not fully agree with the proposal, but I think we can still find a middle ground and move forward.

Analysis

Summary: Polite disagreement looking for a workable compromise
Main theme: Negotiation
Tone: Conciliatory
Intent: Negotiate
Dominant emotion: Controlled tension

Recommended quote

But now let each becalm his troubled breast,
Wash, and partake serene the friendly feast.

Homer, The Odyssey

Why it fits

It cools the temperature without sounding weak. The quote shifts the message away from friction and toward calm, shared ground, and continued conversation.

Example 2

Input

We need to stay focused, make a decision, and keep moving even if the road is rough.

Analysis

Summary: Call for disciplined action under pressure
Main theme: Leadership
Tone: Resolute
Intent: Persuade
Dominant emotion: Focus

Recommended quote

In battle calm he guides the rapid storm,
Wise to resolve, and patient to perform.

Homer, The Odyssey

Why it fits

It is short, memorable, and action-oriented. The line matches a message that asks for composure, judgment, and forward motion at the same time.

Bedrock as the Language Brain

Language detection, rhetorical analysis, quote selection, and translation all go through AWS Bedrock (running Claude Sonnet 4). I tried a local language detector first, but short real-world messages are messy: mixed languages, ticket IDs, URLs, corporate jargon. Bedrock handles that ambiguity much better and keeps the pipeline simpler.

The deterministic part of the system is still retrieval and reranking. But for anything that requires actually understanding language, Bedrock does the heavy lifting.

Reading EPUBs With Zero Dependencies

The corpus loader handles EPUB files using only Python’s standard library, zipfile, xml.etree.ElementTree, and html.parser:

			
def _load_epub(path: Path) -> LoadedDocument:
    with zipfile.ZipFile(path, "r") as archive:
        container_xml = archive.read("META-INF/container.xml")
        container_root = ET.fromstring(container_xml)
        rootfile = container_root.find(".//c:rootfile", CONTAINER_NS)
        opf_root = ET.fromstring(archive.read(opf_path))
        # Parse manifest, follow the spine, extract XHTML chapters
        for chapter_path in spine_paths:
            chapter_html = archive.read(chapter_path).decode("utf-8", errors="ignore")
            text = _extract_epub_text(chapter_html)
            chapters.append(text)

		

An EPUB is really just a ZIP file with XML and HTML inside. The loader opens the ZIP, reads the table of contents (the OPF file), follows the chapter order (the spine), and extracts clean text from each HTML chapter. No external libraries needed, just Python’s built-in tools.

The HashingEmbedder: Deterministic Tests Without Models

Running the full pipeline in tests means loading Sentence Transformers, which means downloading a 90MB model. Instead, there’s a HashingEmbedder that produces fake-but-consistent embeddings using SHA1:

			
class HashingEmbedder:
    def _embed(self, text: str) -> np.ndarray:
        vector = np.zeros(self.dimension, dtype=np.float32)
        for token in self._tokenize(text):
            digest = hashlib.sha1(token.encode("utf-8")).hexdigest()
            bucket = int(digest[:8], 16) % self.dimension
            sign = 1.0 if int(digest[8:10], 16) % 2 == 0 else -1.0
            vector[bucket] += sign
        norm = np.linalg.norm(vector)
        if norm > 0:
            vector /= norm
        return vector

		

The idea: each word gets hashed to a position and a direction in the vector. The same word always produces the same result, so the tests are reproducible. These embeddings don’t understand meaning, “king” and “queen” won’t be close, but the whole pipeline runs exactly the same way, with real numbers flowing through every step. Tests stay fast, offline, and predictable.

Source code

Full source code available in my GitHub repository.

AI-Powered CloudWatch Logs Analysis with Python and Strands Agents

Gonzalo Ayuso — Mon, 11 May 2026 12:24:03 +0000

When you’re debugging production issues at 3 AM, the last thing you want is to scroll through thousands of CloudWatch log entries trying to find that one error. I’ve built a CLI tool that uses AWS Bedrock (Claude Sonnet 4.5) to analyze CloudWatch logs intelligently. You ask questions in natural language, and it gives you insights instead of raw log dumps.

This project is an exploration of combining AWS services with AI agents. Yes, it’s probably over-engineered for simple log queries, but it demonstrates interesting patterns for handling large datasets with parallel AI processing.

The Problem

CloudWatch Logs Insights is powerful, but it has limitations:

You need to know the query syntax
Results are raw data, not insights
Large result sets are overwhelming
Pattern recognition requires manual analysis

What if you could ask: “What errors occurred in the last 2 hours?” and get an intelligent summary instead of 10,000 raw log entries?

Architecture

The tool implements two interaction modes: direct CLI queries and an interactive agent.

Key Components

CloudWatch Insights Query Layer (modules/logs/main.py)

Recursively subdivides time ranges when hitting AWS’s 10,000 result limit
Parses natural language time ranges (“last 2 hours”, “since yesterday”)
Supports custom CloudWatch Insights query syntax

Smart Dataset Routing

Small datasets (2,000 logs): Parallel worker-coordinator pattern
Configurable chunk size and max workers

Worker-Coordinator Pattern Each worker agent analyzes a chunk of logs (2,000 records), then a coordinator agent synthesizes all analyses into a coherent answer. This architecture allows processing 10,000+ log records efficiently while staying within Claude’s context limits.

Interactive Agent (agents/log_agent.py) A specialized agent with access to the analyze_cloudwatch_logs tool, configured with known log groups and time parsing capabilities.

Technology Stack

Python 3.13 with type hints and Pydantic models
boto3 for AWS CloudWatch Logs API
AWS Bedrock (Claude Sonnet 4.5) for AI analysis
Strands Agents for agent orchestration and tool integration
Click for CLI interface

Create your AWS credentials and configure the tool:

# Environment variables
AWS_REGION=eu-central-1
AWS_PROFILE_NAME=your-profile
MAX_CHUNKS_TO_PROCESS=5  # Safety limit for cost control

# Optional: Define known log groups
KNOWN_LOG_GROUPS="/aws/lambda/api,/aws/ecs/backend"

Configure known log groups in settings.py:

class LogGroups(StrEnum):
    """Known CloudWatch log groups for type-safe references."""
    API_LAMBDA = "/aws/lambda/api"
    BACKEND_ECS = "/aws/ecs/backend-service"
    DATABASE_RDS = "/aws/rds/instance/prod/postgresql"

Now you can ask specific questions about your logs:

poetry run python src/cli.py log \
  --group "/aws/lambda/api-handler" \
  --question "What errors occurred?" \
  --start "2025-12-20T10:00:00" \
  --end "2025-12-20T12:00:00"

With custom CloudWatch Insights query:

poetry run python src/cli.py log \
  --group "/aws/lambda/payment" \
  --question "Analyze payment failures" \
  --start "2025-12-20" \
  --query "fields @timestamp, @message, userId | filter @message like /ERROR/"

Using natural language time ranges:

poetry run python src/cli.py log \
  --group "/aws/ecs/backend" \
  --question "What performance issues occurred?" \
  --start "last 2 hours"

The project also allows us to launch an interactive session for exploratory analysis:

poetry run python src/cli.py agent

Example interaction:

============================================================
CloudWatch Logs Analysis Agent
============================================================
Ask me about your CloudWatch logs!

Examples:
  - What errors occurred in /aws/lambda/api in the last hour?
  - Analyze /aws/ecs/backend-service from last 2 hours for memory issues
  - Show me exceptions in /aws/lambda/payment-api since yesterday

Type 'exit', 'quit', or 'q' to quit.
============================================================

> What errors happened in /aws/lambda/api in the last hour?

[Agent analyzes logs and provides intelligent summary]

> Were there any timeouts?

[Agent refines analysis based on context]

The agent mode maintains conversation context and can refine analyses based on follow-up questions.

CloudWatch Insights limits results to 10,000 records per query. The tool automatically subdivides time ranges when hitting this limit:

def query_chunk_recursively(log_group: str, start: datetime, end: datetime,
                           query: str, depth: int = 0) -> list[list[dict]]:
    """
    Queries a time chunk and subdivides it recursively if it hits the result limit.
    Returns all log entries for the given time range.
    """
    status, rows = insights_query(log_group, start=start, end=end,
                                 query=query, limit=MAX_RESULTS_PER_QUERY)

    if len(rows) >= MAX_RESULTS_PER_QUERY:
        # Subdivide in half
        midpoint = start + (end - start) / 2
        first_half = query_chunk_recursively(log_group, start, midpoint, query, depth + 1)
        second_half = query_chunk_recursively(log_group, midpoint, end, query, depth + 1)
        return first_half + second_half

    return rows

This ensures you can analyze any time range without manual intervention, regardless of log volume.

For large datasets, logs are split into chunks and processed in parallel:

def analyze_chunk_with_worker(
    chunk: LogChunk, question: str, log_group: str, global_metadata: dict
) -> ChunkAnalysisResult:
    """
    Analyze a single chunk of logs using a worker agent.
    Each worker gets chunk-specific context and the user's question.
    """
    worker_prompt = WORKER_AGENT_PROMPT.format(
        chunk_index=chunk.chunk_index + 1,
        total_chunks=chunk.total_chunks,
        chunk_size=chunk.chunk_size,
        time_range=chunk.get_time_range_description(),
        question=question,
    )

    worker_agent = create_agent(
        system_prompt=worker_prompt,
        model=Models.CLAUDE_45,
        temperature=0.3,
        read_timeout=WORKER_TIMEOUT_SECONDS,
    )

    chunk_context = {
        "metadata": {...},
        "logs": chunk.logs,
    }

    result = worker_agent(prompt=[
        {"text": f"Question: {question}"},
        {"text": f"Log context: {json.dumps(chunk_context)}"},
        {"text": "Analyze this chunk of logs according to the guidelines in your system prompt."},
    ])

    return ChunkAnalysisResult(...)

Workers run concurrently using ThreadPoolExecutor:

with ThreadPoolExecutor(max_workers=MAX_PARALLEL_WORKERS) as executor:
    future_to_chunk = {
        executor.submit(analyze_chunk_with_worker, chunk, question, log_group, global_metadata): chunk
        for chunk in chunks
    }

    for future in as_completed(future_to_chunk):
        result = future.result()
        chunk_results.append(result)

After workers complete, a coordinator agent synthesizes their analyses:

def consolidate_with_coordinator(
    chunk_results: list[ChunkAnalysisResult],
    question: str,
    log_group: str,
    start: datetime,
    end: datetime,
    total_records: int,
) -> str:
    """
    Use coordinator agent to synthesize chunk analyses into final answer.
    """
    coordinator_context = {
        "metadata": {
            "log_group": log_group,
            "time_range": f"{start.isoformat()} to {end.isoformat()}",
            "total_records": total_records,
            "total_chunks": len(chunk_results),
        },
        "chunk_analyses": [
            {
                "chunk_index": r.chunk_index + 1,
                "time_range": r.chunk_time_range,
                "analysis": r.analysis,
            }
            for r in successful_results
        ],
    }

    result = coordinator(prompt=[
        {"text": f"Original Question: {question}"},
        {"text": f"Chunk Analyses: {json.dumps(coordinator_context)}"},
        {"text": "Synthesize these chunk analyses to answer the user's question."},
    ])

    return str(result)

This pattern allows analyzing datasets far exceeding Claude’s context window while maintaining coherent insights.

The tool supports flexible time specifications:

# modules/logs/time_parser.py
def parse_time_range(time_range: str) -> tuple[datetime, datetime]:
    """
    Parse natural language time ranges:
    - "last 2 hours"
    - "since yesterday"
    - "2025-12-10 to 2025-12-12"
    - "last 7 days"
    """
    # Implementation handles various patterns
    pass

The analyze_cloudwatch_logs tool integrates with Strands agents:

@tool
def analyze_cloudwatch_logs(
    log_group: Annotated[Union[LogGroups, str], "CloudWatch log group name"],
    question: Annotated[str, "Question to answer about the logs"],
    time_range: Annotated[Optional[str], "Time range examples: 'last 2 hours', 'since yesterday'"] = None,
    cloudwatch_sql: Annotated[Optional[str], "CloudWatch Insights query string"] = None,
) -> dict:
    """
    Analyze AWS CloudWatch Logs to answer questions about application behavior.
    Automatically handles large datasets through parallel chunking.
    """
    # Parse time range
    start_dt, end_dt = parse_time_range(time_range or "last 24 hours")

    # Call the existing analysis function
    analysis, metadata = ask_to_log(log_group, question, start_dt, end_dt,
                                   cloudwatch_sql=cloudwatch_sql or DEFAULT_CW_SQL)

    return {
        "status": "success",
        "content": [{"text": f"Analysis for log group '{log_group}':\n\n{analysis}"}],
        "metadata": metadata,
    }

This tool can be composed with other agent tools for more sophisticated workflows.

Each worker agent receives context about its role in the larger analysis:

WORKER_AGENT_PROMPT = """You are a CloudWatch Logs Analysis Worker Agent.

Role: Analyze a specific chunk of logs (part {chunk_index} of {total_chunks})
Time range: {time_range}
Chunk size: {chunk_size} log records

Your task:
1. Analyze this chunk for patterns, errors, anomalies related to: {question}
2. Provide factual observations, not speculation
3. Note timestamps for important events
4. Be concise - a coordinator will synthesize all chunks

Focus on:
- Error messages and stack traces
- Unusual patterns or spikes
- Performance indicators
- User-impacting events

Output format: Concise bullet points with timestamps.
"""

The coordinator synthesizes worker outputs into coherent insights:

COORDINATOR_AGENT_PROMPT = """You are a CloudWatch Logs Coordinator Agent.

Role: Synthesize analyses from {chunks_processed} worker agents
Dataset: {total_records} total log records
Time range: {time_range}

You've received chunk-level analyses. Your task:
1. Identify patterns across all chunks
2. Synthesize a coherent narrative answering the user's question
3. Highlight critical findings
4. Provide actionable insights

Output format:
- Executive summary
- Key findings (chronological if relevant)
- Patterns or trends observed
- Recommendations (if applicable)

Be direct and actionable. Focus on what matters.
"""

Processing large log volumes with AI can get expensive. The tool includes configurable safety limits:

# settings.py
MAX_CHUNKS_TO_PROCESS = int(os.getenv("MAX_CHUNKS_TO_PROCESS", "5"))

# In main.py
if len(chunks) > MAX_CHUNKS_TO_PROCESS:
    error_msg = (
        f"Dataset would generate {len(chunks)} chunks, which exceeds the maximum limit "
        f"of {MAX_CHUNKS_TO_PROCESS} chunks.\n\n"
        f"Options:\n"
        f"  1. Reduce time range to analyze fewer logs\n"
        f"  2. Increase MAX_CHUNKS_TO_PROCESS in settings\n"
        f"  3. Use more specific CloudWatch Insights filters"
    )
    return f"ERROR: {error_msg}", {...}

With default settings (chunk size = 2,000, max chunks = 5), you can analyze up to 10,000 log records per query. Adjust these values based on your budget and requirements.

The project uses Pydantic for all data structures:

# modules/logs/models.py
class LogChunk(BaseModel):
    """Represents a chunk of logs for parallel processing."""
    chunk_index: int
    total_chunks: int
    chunk_size: int
    start_timestamp: str | None
    end_timestamp: str | None
    logs: list[dict[str, str]]

    def get_time_range_description(self) -> str:
        if self.start_timestamp and self.end_timestamp:
            return f"{self.start_timestamp} to {self.end_timestamp}"
        return "Unknown time range"


class ChunkAnalysisResult(BaseModel):
    """Result from a worker agent analyzing a chunk."""
    chunk_index: int
    chunk_time_range: str
    chunk_size: int
    analysis: str
    success: bool = True
    error_message: str | None = None
    processing_time_seconds: float = 0.0

Let’s say you’re investigating a production incident:

# Step 1: Check what happened in the last hour
poetry run python src/cli.py log \
  --group "/aws/lambda/payment-api" \
  --question "What errors occurred?" \
  --start "last 1 hour"

# Output:
# Analysis for log group '/aws/lambda/payment-api' from 2025-12-20T14:00:00 to 2025-12-20T15:00:00:
#
# Key Findings:
# - 47 payment timeout errors between 14:23 and 14:45
# - Errors clustered around Stripe API calls
# - No database connection issues observed
# - Timeout duration: consistently 30 seconds
#
# Pattern: All failures occurred during userId sessions starting with 'eu-'
# suggesting regional routing issue.
#
# [Metadata: 8,432 records, 5 chunks, 23.4s]

The agent identified the pattern (regional issue) and specific time window without you writing complex queries or manually reviewing logs.

This project is deliberately over-engineered. For simple log queries, CloudWatch Insights is sufficient. But building this taught me about:

Managing AI context window limits at scale
Worker-coordinator patterns for parallel processing
Designing tools for agent consumption
Balancing cost vs. capability in AI systems

We can use this tool effectively in scenarios like:

Debugging complex incidents requiring pattern recognition
Onboarding new team members who don’t know your query syntax
Exploratory analysis where you don’t know what you’re looking for
Generating incident reports from raw logs

When NOT to Use This:

Real-time monitoring (use CloudWatch alarms)
Known queries you run repeatedly (use saved Insights queries)
Cost-sensitive environments (AI analysis adds expense)

AI agents transform log analysis from query construction to question asking. Instead of learning CloudWatch Insights syntax, you describe what you want to know. The worker-coordinator pattern demonstrates how to scale AI analysis beyond single-agent context limits.

Is it practical for every use case? No. Is it interesting to build and explore? Absolutely.

The complete implementation is available in my GitHub account.

Removing Clickbait from News Articles with an AI Agent, Python, Strands Agents, and AWS Bedrock

Gonzalo Ayuso — Mon, 04 May 2026 12:15:05 +0000

The web is full of articles that do not want to tell you what happened too soon. The headline hints at something. The first paragraphs add suspense. The useful information is somewhere below the fold, after the cookie banner, the newsletter box, a couple of related links, and enough scrolling to make the advertising model happy.

That is annoying when all we want is the news.

That’s my PoC. A small command-line application that receives the URL of a news article, converts the page into clean Markdown, and asks an AI agent to rewrite it as clear journalism: direct headline, concise lead, short paragraphs, no clickbait.

The idea is simple:

plainnews rewrite "https://example.com/news/article"

The CLI does not scrape the page directly. It gives the URL to a Strands Agent. The agent has one tool, fetch_url_as_markdown, and the model decides when to use it. Once the article is available as Markdown, the agent rewrites it following a focused system prompt.

The architecture

The flow is straightforward:

The important part is the boundary between the agent and the tool. Fetching a web page, removing navigation, and converting HTML into Markdown is deterministic Python code. Deciding how to rewrite the story is the LLM’s job.

This keeps the PoC small and easy to reason about.

Project structure

I like to keep configuration in settings.py. It is a pattern I borrowed years ago from Django and I still use it in small prototypes because it keeps things simple:

src/
  cli.py
  settings.py
  commands/
    rewrite.py
  lib/
    agent.py
    prompts.py
    tools.py
    ui.py
  env/
    local/
      .env.example
tests/

The responsibilities are intentionally small:

src/commands/rewrite.py contains the Click command.
src/lib/tools.py contains the Strands tool and the HTML-to-Markdown pipeline.
src/lib/agent.py wires Strands Agents with AWS Bedrock.
src/lib/prompts.py keeps the editor prompt and the user task prompt.
src/lib/ui.py renders Markdown in the terminal with Rich.

Fetching a URL as Markdown

The agent only gets one tool. It fetches the URL, removes noisy page elements, selects the main content, converts it to Markdown, and truncates the result to 100K characters:

			
@tool
def fetch_url_as_markdown(url: str) -> str:
    """
    Fetch an HTTP or HTTPS URL, remove navigation, ads, scripts and layout noise,
    extract the main article content, convert it to Markdown, and return up to
    100K characters of clean text.
    Use this tool when the user pastes a URL or asks you to analyze a web page.
    """
    return fetch_url_as_markdown_impl(url)
def clean_html_to_markdown(html: str, *, max_chars: int = 100_000) -> str:
    soup = BeautifulSoup(html, "html.parser")
    for selector in NOISY_SELECTORS:
        for tag in soup.select(selector):
            tag.decompose()
    content = soup.find("main") or soup.find("article") or soup.body
    if content is None:
        return ""
    markdown = md(str(content), heading_style="ATX", bullets="-", strip=["a"])
    markdown = normalize_markdown(markdown)
    if len(markdown) > max_chars:
        return markdown[:max_chars].rstrip() + "\n\n[Content truncated]"
    return markdown

		

I am not trying to build a perfect browser engine here. This is a PoC. The goal is to get enough readable article content for the agent to work with. For many news pages, removing scripts, navigation, cookie boxes, newsletter blocks, related links and advertising containers is enough.

The agent

The agent uses Claude on AWS Bedrock through Strands Agents:

			
def create_agent(*, settings: Settings) -> Agent:
    boto_session = create_boto_session(settings)
    return Agent(
        model=BedrockModel(
            boto_session=boto_session,
            model_id=settings.resolved_bedrock_model_id,
        ),
        tools=[fetch_url_as_markdown],
        system_prompt=SYSTEM_PROMPT,
    )

		

The system prompt is the editorial policy. It tells the model to preserve only facts supported by the fetched article, answer in the requested output language, put the most important information first, remove suspense and filler, and write in a neutral tone.

The output format is Markdown:

a direct H1 headline
a concise lead paragraph
short factual paragraphs
a final What changed section, translated to the requested output language, explaining what noise was removed

That last section is useful during development. It gives us a quick sanity check: did the model actually remove clickbait, or did it just paraphrase the article?

The CLI

The command is intentionally small:

			
@click.command(name="rewrite")
@click.argument("url")
@runtime_options
def rewrite_command(
    url: str,
    aws_profile: str | None,
    region: str | None,
    model: str | None,
    language: str,
) -> None:
    if not is_supported_url(url):
        raise click.ClickException("URL must start with http:// or https://")
    settings = resolve_settings(
        aws_profile=aws_profile,
        aws_region=region,
        bedrock_model_id=model,
    )
    agent = create_agent(settings=settings)
    result = agent(build_rewrite_prompt(url, language=language))
    print_result("PlainNews", str(result))

		

The CLI validates the URL, creates the agent, sends the URL in the prompt, and renders the final Markdown with Rich.

The tool is not called manually from the command. That is the point of this PoC: the URL is part of the task, and the agent decides to call fetch_url_as_markdown because the tool description says it should be used when the user pastes a URL or asks to analyze a web page.

Usage

Run the command:

poetry run plainnews rewrite "https://example.com/news/article"

By default, PlainNews writes the rewritten article in English. You can choose a different output language with --language:

poetry run plainnews rewrite "https://example.com/news/article" --language Spanish

The output is rendered as Markdown in the terminal.

Example terminal output:

Tech stack

Python with Poetry
Strands Agents for tool-based agent orchestration
AWS Bedrock for the LLM runtime
BeautifulSoup for HTML cleanup
markdownify for HTML-to-Markdown conversion
Click for the command-line interface
Rich for Markdown terminal rendering
pytest for tests

A couple of notes

This is not a product and it is not a universal paywall remover. It is a small agentic workflow for a very specific frustration: articles that make readers work too hard to understand the basic facts.

Even in this small version, the pattern is useful: deterministic Python code prepares clean context, and the AI agent performs the editorial rewrite with a tight prompt.

And that’s all. Full source code available on GitHub.

Production-Ready Logging with AWS CloudWatch for Python applications

Gonzalo Ayuso — Mon, 27 Apr 2026 12:13:22 +0000

Today we’re going to build a production-grade logging system for Python applications using. We’re going to use CloudWatch Agent with it’s auto_removal feature to automatically delete log files after they’ve been uploaded to CloudWatch Logs.

This architectural constraint requires careful design of your logging pipeline.

The solution uses a dual-formatter approach:

Console output: Human-readable format for docker compose logs
File output: Structured JSON for CloudWatch with hourly rotation
CloudWatch Agent: Reads rotated files and automatically deletes them after upload

Flask App
    ↓
Logging System (dual formatters)
    ├─→ Console Handler → Human-readable + extras
    └─→ File Handler → JSON (hourly rotation)
            ↓
        Rotated files (app.log.YYYY-MM-DD_HH)
            ↓
        CloudWatch Agent (auto_removal: true)
            ↓
        AWS CloudWatch Logs

The logging system uses two custom formatters to serve different purposes:

from datetime import datetime
from logging.handlers import TimedRotatingFileHandler
from pythonjsonlogger.json import JsonFormatter
import logging

class CloudWatchJsonFormatter(JsonFormatter):
    """JSON formatter for CloudWatch logs with custom metadata fields."""

    def __init__(self, app: str, process: str, *args, **kwargs):
        self.app = app
        self.process = process
        super().__init__(*args, **kwargs)

    def add_fields(self, log_record, record, message_dict):
        """Add CloudWatch-specific fields to log record."""
        log_record['@timestamp'] = datetime.fromtimestamp(record.created).isoformat()
        log_record['level'] = record.levelname
        log_record['app'] = self.app
        log_record['logger'] = record.name
        log_record['process'] = self.process
        super(CloudWatchJsonFormatter, self).add_fields(log_record, record, message_dict)


class ConsoleFormatter(logging.Formatter):
    """Console formatter that includes extra fields."""

    RESERVED_ATTRS = {
        'name', 'msg', 'args', 'created', 'filename', 'funcName', 'levelname',
        'levelno', 'lineno', 'module', 'msecs', 'message', 'pathname', 'process',
        'processName', 'relativeCreated', 'thread', 'threadName', 'exc_info',
        'exc_text', 'stack_info', 'asctime'
    }

    def format(self, record):
        base_message = super().format(record)

        # Extract extra fields
        extras = {
            key: value
            for key, value in record.__dict__.items()
            if key not in self.RESERVED_ATTRS
        }

        if extras:
            extras_str = ' '.join(f'{k}={v}' for k, v in extras.items())
            return f'{base_message} | {extras_str}'

        return base_message

The ConsoleFormatter automatically appends extra fields to the log message, making debugging easier. The CloudWatchJsonFormatter creates structured JSON logs with CloudWatch-specific metadata.

The key to making auto_removal work is using TimedRotatingFileHandler with hourly rotation:

from typing import Literal, Union
from pathlib import Path

def setup_logging(
env: Literal[‘local’, ‘production’],
app: str,
log_path: Union[Path, str],
process: str = ‘main’,
log_level: str = ‘INFO’
) -> None:
“””Configure logging with environment-specific settings.”””
if env == ‘local’:
logging.basicConfig(
format=’%(asctime)s [%(levelname)s] %(message)s’,
level=log_level,
datefmt=’%d/%m/%Y %X’
)
else:
# Console handler with human-readable format
console_handler = logging.StreamHandler()
console_formatter = ConsoleFormatter(
fmt=’%(asctime)s [%(levelname)s] %(name)s – %(message)s’,
datefmt=’%Y-%m-%d %H:%M:%S’
)
console_handler.setFormatter(console_formatter)

    # JSON formatter for file (CloudWatch)
    json_formatter = CloudWatchJsonFormatter(
        app=app,
        process=process,
        fmt='%(levelname)s %(name)s %(message)s'
    )

    # File handler with hourly rotation
    file_handler = TimedRotatingFileHandler(
        log_path,
        when='H',        # Hourly rotation
        interval=1,      # Every 1 hour
        backupCount=2,   # Keep 2 backups
        encoding='utf-8'
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(json_formatter)

    logging.basicConfig(
        level=log_level,
        handlers=[console_handler, file_handler]
    )

Why hourly rotation? CloudWatch Agent’s auto_removal only deletes complete files. With daily rotation, you’d have up to 24 hours of logs accumulating. Hourly rotation minimizes disk usage to just 1-2 hours of logs at any time.

Important: Do not use minute-level rotation (when='M'). Fast rotation intervals cause timing issues where CloudWatch Agent cannot properly track file inodes during rotation, leading to log loss or incorrect file deletion. AWS documentation recommends hourly or longer rotation intervals for reliable auto_removal behavior.

Using the logger is straightforward. Extra fields are automatically handled:

import logging
from flask import Flask
from lib.logger import setup_logging
from settings import APP, PROCESS, LOG_PATH, ENVIRONMENT

app = Flask(__name__)
logger = logging.getLogger(__name__)

setup_logging(
    env=ENVIRONMENT,
    app=APP,
    process=PROCESS,
    log_path=LOG_PATH
)

@app.get("/")
def health():
    logger.info("GET /", extra=dict(
        user_id=123,
        response_time_ms=45
    ))
    return {'status': 'ok'}

Console output:

2025-12-13 12:30:45 [INFO] app - GET / | user_id=123 response_time_ms=45

JSON output (CloudWatch):

{
  "@timestamp": "2025-12-13T12:30:45.123456",
  "level": "INFO",
  "logger": "app",
  "app": "cw_demo",
  "process": "cw_demo",
  "message": "GET /",
  "user_id": 123,
  "response_time_ms": 45
}

The CloudWatch Agent uses auto_removal to delete rotated files automatically:

{
  "agent": {
    "debug": false
  },
  "logs": {
    "logs_collected": {
      "files": {
        "collect_list": [
          {
            "file_path": "/logs/*.log*",
            "log_group_name": "${LOG_GROUP_NAME}",
            "log_stream_name": "{hostname}",
            "auto_removal": true
          }
        ]
      }
    }
  }
}

The wildcard pattern /logs/*.log* matches any log file and its rotations (e.g., app.log, cw.log, worker.log and their rotated versions like app.log.2025-12-13_12). This allows different applications to use different log file names based on their app_id.

The LOG_GROUP_NAME environment variable is injected at runtime by the entrypoint script. If not provided, it defaults to /app/logs.

The application runs alongside CloudWatch Agent with a shared volume:

services:
  api:
    build:
      context: .
    volumes:
      - logs_volume:/src/logs
    environment:
      - ENVIRONMENT=production
      - PROCESS_ID=api
    ports:
      - 5000:5000
    command: gunicorn -w 1 app:app -b 0.0.0.0:5000 --timeout 180

  cloudwatch-agent:
    build:
      context: .docker/cw
    volumes:
      - logs_volume:/logs
    environment:
      - LOG_GROUP_NAME=/mi-proyecto/app  # Optional: defaults to /app/logs if not set

volumes:
  logs_volume:

The CloudWatch Agent container’s entrypoint script handles the LOG_GROUP_NAME variable with a default value:

#!/bin/sh
set -e

# Set default value for LOG_GROUP_NAME if not provided
LOG_GROUP_NAME=${LOG_GROUP_NAME:-/app/logs}

# Replace LOG_GROUP_NAME environment variable in the config file
sed "s|\${LOG_GROUP_NAME}|${LOG_GROUP_NAME}|g" \
    /opt/aws/amazon-cloudwatch-agent/bin/config.template.json > /opt/aws/amazon-cloudwatch-agent/bin/default_linux_config.json

echo "CloudWatch Agent starting with LOG_GROUP_NAME=${LOG_GROUP_NAME}"

# Start the CloudWatch Agent
exec /opt/aws/amazon-cloudwatch-agent/bin/start-amazon-cloudwatch-agent

This ensures the agent always has a valid log group name, even if the environment variable is not explicitly set.

When using CloudWatch Insights, remember to use the actual field names from your JSON structure:

fields @timestamp, level, logger, message, user_id, response_time_ms
| filter level = "ERROR"
| sort @timestamp desc
| limit 100

This logging architecture uses sidecar patterns to decouple application logic from logging concerns, ensuring robust, production-ready logging with minimal disk usage and automatic log management via AWS CloudWatch.

full code in my GitHub account.

What if you could ask questions to any GitHub repository? Building a repository-aware AI agent with Python, Strands Agents, and Bedrock

Gonzalo Ayuso — Mon, 06 Apr 2026 14:19:41 +0000

Sometimes we land on an unfamiliar GitHub repository and the first problem is not writing code. The real problem is understanding the project fast enough. Is this a REST API? Where are the entrypoints? How is the application wired? Are there obvious risks in the codebase? If the repository is big enough, answering those questions manually is slow and boring.

That’s just my PoC. An interactive command-line application that can inspect any public GitHub repository and answer questions about it.

I have the feeling this workflow should exist natively on GitHub. Once repositories become large enough, being able to ask architecture, audit, or API questions feels like a natural evolution of code search and Copilot. Maybe the reason it does not exist yet is cost, scope, or product complexity. In the meantime, a CLI-first open source approach feels like a good place to start: simple, scriptable, hackable, and based on bring-your-own-model credentials so each user keeps control of their own usage and billing.

The idea is simple. We give a GitHub repository to a CLI application. The CLI creates a local checkout, exposes a small set of repository-aware tools to a Strands Agent, and lets the agent inspect the project with AWS Bedrock. Because the agent can list directories, search code and read files, we can ask practical questions such as:

Explain how the project works
Audit the codebase looking for risks
List the API endpoints
Describe the execution flow of a specific module

This is not a vector database project and it is not a RAG pipeline. It is a much simpler approach. We let the agent explore the repository directly, file by file, using tools.

The architecture

The flow is straightforward:

The user calls the CLI with a GitHub repository.
The repository is cloned into a local cache.
A Strands Agent is created with a Bedrock model.
The agent receives a system prompt plus four tools: get_directory_tree, list_directory, search_code and read_file.
The agent inspects the repository and returns the final answer in Markdown.

This is enough for a surprising number of use cases. If the system prompt is focused on architecture, the answer becomes an explanation. If the prompt is focused on risk, the answer becomes a code audit. If the prompt is focused on HTTP routes, the answer becomes an API inventory.

Project structure

I like to keep configuration in settings.py. It is a pattern I borrowed years ago from Django and I still use it in small prototypes because it keeps things simple:

src/
└── github_kb/
    ├── cli.py
    ├── settings.py
    ├── commands/
    │   ├── ask.py
    │   ├── audit.py
    │   ├── chat.py
    │   ├── endpoints.py
    │   └── explain.py
    ├── lib/
    │   ├── agent.py
    │   ├── github.py
    │   ├── models.py
    │   ├── prompts.py
    │   ├── repository.py
    │   └── ui.py
    └── env/
        └── local/
            └── .env.example

The responsibilities are small and explicit:

github_kb/commands/ contains the Click commands.
github_kb/lib/github.py resolves the GitHub repository and manages the local checkout.
github_kb/lib/repository.py contains the repository exploration logic used by the agent tools.
github_kb/lib/agent.py wires Strands Agents with AWS Bedrock.
github_kb/lib/prompts.py keeps the system prompt and the task-specific prompts in one place.

Why this works

Large repositories are difficult because we rarely need the whole repository at once. We normally need a guided exploration strategy. A tree view helps us identify the shape of the project. Search helps us jump to the interesting files. Reading files gives us the final confirmation.

That sequence maps very well to tool-based agents.

Instead of trying to send the whole repository in one prompt, the model can progressively inspect only the relevant parts. It is cheaper, easier to reason about, and much closer to how we inspect an unknown codebase ourselves.

Install

The intended installation flow is:

pipx install github-kb

Quick start

The happy path should look like this:

aws sso login --profile sandbox
AWS_PROFILE=sandbox AWS_REGION=us-west-2 github-kb doctor
AWS_PROFILE=sandbox AWS_REGION=us-west-2 github-kb chat gonzalo123/autofix

The CLI is designed to work out of the box with the standard AWS credential chain. That means it can use:

AWS_PROFILE
AWS_REGION
aws sso login
regular access keys if they are already configured in the environment

By default, github-kb uses global.anthropic.claude-sonnet-4-6 unless BEDROCK_MODEL_ID or --model says otherwise.

You can also override the runtime explicitly with CLI flags such as --aws-profile, --region, and --model.

Usage

Now we can ask questions:

github-kb ask gonzalo123/autofix "How does the automated fix flow work?"
github-kb chat gonzalo123/autofix
github-kb explain gonzalo123/autofix --topic architecture
github-kb audit gonzalo123/autofix --focus github
github-kb endpoints gonzalo123/autofix
github-kb doctor

If we want to keep the same conversation alive across multiple questions in one terminal session:

github-kb chat gonzalo123/autofix

It also accepts full GitHub URLs:

github-kb ask https://github.com/gonzalo123/autofix "Where is the application bootstrapped?"

If we want to refresh the local cache:

github-kb audit gonzalo123/autofix --refresh

We can also pass the AWS runtime explicitly:

github-kb chat gonzalo123/autofix --aws-profile sandbox --region eu-central-1
github-kb ask gonzalo123/autofix "Explain the architecture" --model global.anthropic.claude-sonnet-4-6

Demo screenshots

Here are a few real screenshots generated against one of my own repositories, gonzalo123/autofix.

The screenshots below are embedded as PNG files:

`explain`

`endpoints`

`audit`

A couple of notes

This is still a PoC. The goal is not to build a perfect repository analysis platform. The goal is to validate a simple idea: an agent with a tiny set of well-chosen tools can already be useful for code understanding.

There are several obvious next steps:

add more repository-aware tools
persist analysis sessions
summarize previous findings before starting a new question
support GitHub authentication for private repositories
add specialized prompts for security reviews or framework-specific inspections

Even in its current state, it is already a nice example of how tool-based agents can help with a very real developer problem.

Full code in my github

Exposing a REST API Through MCP: Turning Any API into an AI Tool

Gonzalo Ayuso — Mon, 23 Mar 2026 13:34:54 +0000

Most organizations already have REST APIs. They power internal dashboards, connect microservices, expose data to mobile apps. They work. But now you want an AI agent to use those same services, and agents don’t speak REST. They speak MCP. Do you rewrite everything? No. You build a thin adapter layer that translates between the two protocols, and your existing API stays untouched.

That’s exactly what this project does. We take a standard Flask REST API and wrap it with an MCP server using FastMCP. Any MCP-compatible client, Claude Code, Cursor, Windsurf, can discover and call the API endpoints as tools, without knowing there’s a REST layer underneath.

The REST API

The API is a standard Flask application with CRUD endpoints for managing notes. Nothing special here, just the kind of REST service you’d find in any organization:

			
notes_bp = Blueprint("notes", __name__)
@notes_bp.route("/api/notes", methods=["GET"])
def list_notes():
    return jsonify(get_all_notes())
@notes_bp.route("/api/notes/", methods=["GET"])
def read_note(note_id: int):
    note = get_note(note_id)
    if note is None:
        return jsonify({"error": "Note not found"}), 404
    return jsonify(note)
@notes_bp.route("/api/notes", methods=["POST"])
def add_note():
    data = request.get_json()
    if not data or "title" not in data:
        return jsonify({"error": "title is required"}), 400
    note = create_note(title=data["title"], body=data.get("body", ""))
    return jsonify(note), 201
@notes_bp.route("/api/notes/", methods=["PUT"])
def edit_note(note_id: int):
    data = request.get_json()
    note = update_note(note_id, title=data.get("title"), body=data.get("body"))
    if note is None:
        return jsonify({"error": "Note not found"}), 404
    return jsonify(note)
@notes_bp.route("/api/notes/", methods=["DELETE"])
def remove_note(note_id: int):
    if delete_note(note_id):
        return jsonify({"status": "deleted"})
    return jsonify({"error": "Note not found"}), 404

		

Five endpoints: list, get, create, update, delete. The data store is an in-memory dictionary for simplicity, but in a real scenario this would be your existing database, your internal service, your legacy system. The point is that the REST API already exists and works. We don’t want to change it.

The MCP server

This is the core of the project. The MCP server uses FastMCP to expose each REST endpoint as an MCP tool. It uses requests to call the Flask API over HTTP:

			
import requests
from mcp.server.fastmcp import FastMCP
from settings import API_BASE_URL
mcp = FastMCP(name="notes-api")
BASE = API_BASE_URL
@mcp.tool()
def list_notes() -> str:
    """List all notes stored in the system.
    Returns a JSON array of note objects, each containing:
    id, title, body, and created_at fields.
    """
    response = requests.get(f"{BASE}/api/notes")
    return response.text
@mcp.tool()
def get_note(note_id: int) -> str:
    """Get a single note by its ID.
    Returns the note object with id, title, body, and created_at fields.
    Returns an error if the note is not found.
    """
    response = requests.get(f"{BASE}/api/notes/{note_id}")
    return response.text
@mcp.tool()
def create_note(title: str, body: str = "") -> str:
    """Create a new note.
    Args:
        title: The title of the note (required).
        body: The body content of the note (optional, defaults to empty string).
    Returns the created note object with its assigned id.
    """
    response = requests.post(
        f"{BASE}/api/notes",
        json={"title": title, "body": body},
    )
    return response.text
@mcp.tool()
def update_note(note_id: int, title: str = "", body: str = "") -> str:
    """Update an existing note by its ID.
    Args:
        note_id: The ID of the note to update.
        title: New title for the note (optional, send empty string to keep current).
        body: New body for the note (optional, send empty string to keep current).
    Returns the updated note object, or an error if not found.
    """
    payload = {}
    if title:
        payload["title"] = title
    if body:
        payload["body"] = body
    response = requests.put(
        f"{BASE}/api/notes/{note_id}",
        json=payload,
    )
    return response.text
@mcp.tool()
def delete_note(note_id: int) -> str:
    """Delete a note by its ID.
    Args:
        note_id: The ID of the note to delete.
    Returns a status confirmation or an error if the note is not found.
    """
    response = requests.delete(f"{BASE}/api/notes/{note_id}")
    return response.text
if __name__ == "__main__":
    mcp.run()

		

Each @mcp.tool() maps to one REST endpoint. The decorator extracts parameter types from the function signature to build the JSON schema that MCP clients use to understand what parameters to send. The docstring becomes the tool description that the AI agent reads to decide when and how to call each tool. When you run python src/server/main.py, it starts listening on stdio for MCP requests.

The pattern is straightforward: receive the MCP call, translate it into an HTTP request, forward it to the REST API, and return the response. The MCP server knows nothing about the business logic. The REST API knows nothing about MCP. Each side does its job.

The adapter pattern

This is the Adapter Pattern applied at the protocol level. The MCP server adapts the REST interface into the MCP protocol. The REST API doesn’t need to change. The MCP client doesn’t need to know it’s talking to a REST service. The adapter handles the translation:

The MCP client calls create_note("Meeting notes", "Discussed Q3 roadmap"). The MCP server translates this into a POST /api/notes with a JSON body. The Flask API processes it, creates the note, and returns the result. The MCP server passes the response back to the client. The agent sees a tool that creates notes. It doesn’t know or care that there’s an HTTP call in between.

Configuration

To use the MCP server from Claude Code, create a .mcp.json file in your project root:

			
{
  "mcpServers": {
    "notes-api": {
      "command": "/path/to/venv/bin/python",
      "args": ["/path/to/src/server/main.py"]
    }
  }
}

		

Claude Code reads this file, launches the MCP server as a subprocess, performs the MCP handshake, and discovers the five tools automatically. The same server works with Cursor, Windsurf, VS Code with Copilot, or any other MCP-compatible client, just point it to the same Python script.

Running it

First, install dependencies:

poetry install

Start the Flask API in one terminal:

make api

You can verify it works with curl:

curl -X POST http://127.0.0.1:5000/api/notes \
  -H "Content-Type: application/json" \
  -d '{"title": "First note", "body": "Hello from REST"}'

curl http://127.0.0.1:5000/api/notes

With the API running and .mcp.json in place, open Claude Code in the project directory. It discovers the notes-api MCP server and makes all five tools available. You can ask things like “Create a note about the deployment we did today” or “List all my notes” and the agent calls the MCP tools, which call your REST API, automatically.

Taking it further

This POC uses a simple notes API, but the same pattern works with any existing REST service. Your internal APIs, third-party integrations, legacy systems, anything with HTTP endpoints can be wrapped with a thin MCP layer. The REST API stays unchanged, the MCP server handles the translation, and suddenly your existing services become tools that any AI agent can use.

You could also add authentication headers in the MCP server (forwarding API keys or tokens to the REST API), error handling with retry logic, or caching for read-heavy endpoints. The adapter layer is the right place for these cross-cutting concerns.

And that’s all. With a thin MCP adapter on top of any REST API, your existing services become tools that any AI agent can discover and use. The REST API stays unchanged, the MCP server handles the protocol translation, and the standard connects them. Build the adapter once, use it from any MCP client.

Full code in my github account.

AI Eurobeat Producer: Generating Music in Real-Time with AI Agents, Python, and MIDI

Gonzalo Ayuso — Mon, 09 Mar 2026 13:13:39 +0000

What if you could describe the music you want to hear and have an AI produce it in real-time, sending MIDI notes directly to your DAW? That’s exactly what I built: a Python application that uses AI agents to generate Eurobeat and 90s techno patterns, outputting them as live MIDI to Akai’s MPC Beats.

I’m not a musician. I enjoy playing guitar from time to time, but I have zero experience with music production software. However, I’m gifted myself a Akai MPK mini Plus MIDI controller, which has 8 knobs and 8 pads, and I experimented with using it to control a music generation agent. No idea what I’m doing, but it’s fun.

As Akai MIDI controller can be connected to a laptop, and there I’ve got Python, this saturday morning I decided to build a simple prototype that connects an AI agent to MIDI output. The idea is simple. You write a prompt like “Energetic eurobeat in Am, Daft Punk style”, and an AI agent powered by Claude on AWS Bedrock generates patterns for 8 tracks: two drum kits, bass, rhodes, pluck, pad, and a lead melody. The patterns are sent as MIDI messages to MPC Beats, where each track is routed to a different virtual instrument. You can then modify the music live by writing new instructions, and use the physical knobs and pads on an Akai MPK Mini Plus to mute/unmute tracks, regenerate patterns, or reset the session.

I’m using the MPC Beats because it’s free and has a simple MIDI setup, but in theory this could work with any DAW that accepts MIDI input. The whole system is built in Python using Strands Agents for the AI orchestration, mido + python-rtmidi for MIDI I/O, and Rich for the terminal UI.

The Architecture

The flow is straightforward:

Project Structure

src/
  settings.py           # Configuration: BPM, tracks, MIDI devices
  cli.py                # Click CLI entry point
  commands/play.py      # Main play command
  agent/
    prompts.py          # System prompts for the AI producer
    tools.py            # PatternStore + @tool functions
    factory.py          # Agent creation
  midi/
    device.py           # MIDI device detection
    melody_player.py    # Threaded melody loop player
    drum_player.py      # Threaded drum loop player
  session/
    state.py            # State machine (IDLE/GENERATING/PLAYING)
    session.py          # Session orchestrator
  ui/
    menu.py             # Interactive terminal menu

Configuration

Everything starts with settings.py. The MIDI devices and AWS region are loaded from environment variables, while the musical parameters are defined as constants:

			
BPM = 122
BAR_DURATION = round((60 / BPM) * 4, 3)
LOOP_BARS = 4
LOOP_DURATION = round(BAR_DURATION * LOOP_BARS, 3)
TRACKS = {
    1: {"name": "Drums",         "channel": 0, "type": "drums"},
    2: {"name": "Drums Detroit", "channel": 1, "type": "drums"},
    3: {"name": "Rhodes",        "channel": 2, "type": "melody"},
    4: {"name": "Pluck",         "channel": 3, "type": "melody"},
    5: {"name": "Bass",          "channel": 4, "type": "melody"},
    6: {"name": "Org Bass",      "channel": 5, "type": "melody"},
    7: {"name": "Pad",           "channel": 6, "type": "melody"},
    8: {"name": "Lead",          "channel": 7, "type": "melody"},
}

		

Each track maps to a MIDI channel. Tracks 1-2 are drum kits (offset-based timing), tracks 3-8 are melodic instruments (duration-based timing). The MPC Beats “House Template” provides the virtual instruments: a Classic drum kit, a Detroit percussion kit, Electric Rhodes, Tube Pluck, Bassline, Organ Bass, Tube Pad, and an Instant Go lead synth.

The Bridge Between AI and MIDI: PatternStore and Tools

The core of the system is the PatternStore, a simple shared store where the AI writes patterns and the MIDI players read them:

			
class PatternStore:
    def __init__(self):
        self._patterns: dict[int, list] = {}
    def set(self, track_id: int, pattern: list) -> None:
        self._patterns[track_id] = pattern
    def get(self, track_id: int) -> list | None:
        return self._patterns.get(track_id)
    def clear(self) -> None:
        self._patterns.clear()

		

The Strands @tool functions are created via a factory that closes over the store:

			
def create_tools(store: PatternStore) -> list:
    @tool
    def set_melody_pattern(track_id: int, pattern: str) -> str:
        """Define a melodic line for a specific track."""
        data = json.loads(pattern)
        store.set(track_id, data)
        total = sum(n["duration"] for n in data)
        name = TRACKS[track_id]["name"]
        return f"OK - {name}: {len(data)} notes, total duration {total:.3f}s"
    @tool
    def set_drum_pattern(track_id: int, pattern: str) -> str:
        """Define a drum pattern for a specific drum track."""
        data = json.loads(pattern)
        store.set(track_id, data)
        name = TRACKS[track_id]["name"]
        return f"OK - {name}: {len(data)} hits"
    return [set_drum_pattern, set_melody_pattern]

		

A melody pattern is a JSON array of {note, duration, velocity} objects where the sum of durations must equal LOOP_DURATION (4 bars). A drum pattern uses {note, velocity, offset} where offset is the time in seconds from the loop start. The note value -1 represents silence, which is crucial for creating space in the arrangement.

The Agent

The agent is a Strands Agent using Claude Sonnet on AWS Bedrock. The system prompt is heavily detailed with music production instructions: frequency ranges for each track, velocity guidelines, and structural rules. The key instruction is “less is more” – not all tracks should play notes all the time:

			
def create_agent(store: PatternStore) -> Agent:
    return Agent(
        model=BedrockModel(
            model_id=Models.CLAUDE_SONNET,
            region_name=AWS_REGION,
        ),
        tools=create_tools(store),
        system_prompt=SYSTEM_PROMPT,
        callback_handler=None,
    )

		

There are two agents: one for initial generation (calls all 8 tools) and one for live modifications (only modifies the tracks that need to change). A third, lighter agent using Haiku generates the menu suggestions to keep latency and cost low.

MIDI Players

Two player classes handle the actual MIDI output. The MelodyLoopPlayer iterates through note events with durations:

			
def _loop(self, melody: list):
    while not self.stop_event.is_set():
        current = self.store.get(self.track_id) or melody
        for ev in current:
            if self.stop_event.is_set():
                break
            note = ev["note"]
            vel = ev.get("velocity", 80)
            if note >= 0:
                self._send("note_on", note=note, velocity=vel, channel=self.channel)
            deadline = time.time() + ev["duration"]
            while not self.stop_event.is_set() and time.time() < deadline:
                time.sleep(0.02)
            if note >= 0:
                self._send("note_off", note=note, velocity=0, channel=self.channel)

		

The DrumLoopPlayer uses offset-based timing instead, scheduling hits at specific points within the loop. Both players read from the PatternStore on each loop iteration, which enables hot-swapping patterns during live modifications.

The Session

The Session class orchestrates everything. It manages the state machine (IDLE -> GENERATING -> PLAYING), owns the PatternStore, creates the agents, and handles MIDI input from the controller:

			
class Session:
    def __init__(self):
        self.state = State.IDLE
        self.store = PatternStore()
        self.agent = create_agent(self.store)
        self.live_agent = create_live_agent(self.store)
        self._agent_busy = threading.Lock()

		

When generation completes, playback starts with a progressive intro – tracks are unmuted one by one with a 2-bar delay between each, creating a build-up effect:

			
def _start_playback(self):
    self.state = State.PLAYING
    for tid in TRACKS:
        self.players[tid].muted = True
        self.players[tid].start(patterns[tid])
    intro_delay = BAR_DURATION * 2
    for i, tid in enumerate(INTRO_ORDER):
        timer = threading.Timer(intro_delay * i, self._unmute_track, args=(tid,))
        timer.start()

		

How It Works

Run python cli.py play
The app detects your MPK Mini Plus and shows a menu with AI-generated suggestions
Select a suggestion or write your own prompt
The AI generates 8 track patterns (takes a few seconds)
Playback begins with a progressive build-up
Write new instructions to modify the music live
Use knobs K1-K8 to mute/unmute individual tracks
PAD 1 regenerates with the same prompt, PAD 2 resets everything

Tech Stack

Python 3.13 with Poetry
Strands Agents for AI agent orchestration
AWS Bedrock (Claude Sonnet + Haiku) for pattern generation
mido + python-rtmidi for MIDI I/O
Akai MPK Mini Plus as MIDI controller
MPC Beats as the DAW/sound engine
Rich for terminal UI
Click for CLI

And that’s all. Full source code available on GitHub.

Predicting the future: time series forecasting with AI Agents and Amazon Chronos-Bolt

Gonzalo Ayuso — Mon, 02 Mar 2026 13:45:14 +0000

Predicting the future is something we all try to do. Whether it’s energy consumption, sensor readings, or production metrics, having a reliable forecast helps us make better decisions. The problem is that building a good forecasting model traditionally requires deep statistical knowledge, and a lot of tuning. What if we could just hand our data to an AI agent and ask “what’s going to happen next”?

That’s exactly what this project does. It combines Strands Agents with Amazon Chronos-Bolt, a foundation model for time series forecasting available on AWS Bedrock Marketplace, to create an AI agent that can forecast any numerical time series through natural language.

The architecture

The idea is simple. We have a Strands Agent powered by Claude (via AWS Bedrock) that understands natural language. When the user asks for a forecast, the agent calls a custom tool that invokes Chronos-Bolt to generate predictions. The agent then interprets the results and explains them in plain language.

The key here is that the agent doesn’t just return raw numbers. It understands the context, explains trends, and presents the confidence intervals in a way that makes sense.

The forecast tool

The tool is defined using the @tool decorator from Strands. This decorator turns a regular Python function into something the agent can discover and invoke on its own:

			
@tool
def forecast_time_series(
    values: Annotated[
        list[float],
        "Historical time series values in chronological order. "
        "Values should be evenly spaced (e.g., hourly, daily). Minimum 10 values.",
    ],
    prediction_length: Annotated[
        int,
        "Number of future steps to predict. "
        "Uses the same time unit as the input data.",
    ],
    quantile_levels: Annotated[
        Optional[list[float]],
        "Quantile levels for confidence intervals. Default: [0.1, 0.5, 0.9]. "
        "0.5 is the median forecast, 0.1 and 0.9 define the 80% confidence band.",
    ] = None,
) -> dict:

		

The Annotated type hints serve a dual purpose: they validate types at runtime and provide descriptions that the LLM reads to understand how to use the tool. This means the agent knows it needs a list of floats, a prediction length, and optionally custom quantile levels, all from the type annotations alone.

The tool validates the input (minimum 10 values, maximum 50,000, prediction length between 1 and 1,000), filters out NaN values, and then calls the Chronos-Bolt client:

			
result = invoke_chronos(
    values=clean_values,
    prediction_length=prediction_length,
    quantile_levels=quantile_levels,
)
return {
    "status": "success",
    "content": [{"text": "\n".join(summary_lines)}],
    "metadata": {
        "quantiles": result.quantiles,
        "prediction_length": result.prediction_length,
        "history_length": result.history_length,
    },
}

		

The response includes both a human-readable summary (in content) and the raw quantile data (in metadata), so the agent can reference exact numbers when explaining the forecast.

The Chronos-Bolt client

Chronos-Bolt is accessed through the Bedrock runtime API. The client sends the historical values and receives predictions at different quantile levels:

			
def invoke_chronos(
    values: list[float],
    prediction_length: int,
    quantile_levels: list[float] | None = None,
) -> ForecastResult:
    client = _get_bedrock_runtime_client()
    payload = {
        "inputs": [{"target": values}],
        "parameters": {
            "prediction_length": prediction_length,
            "quantile_levels": quantiles,
        },
    }
    response = client.invoke_model(
        modelId=CHRONOS_ENDPOINT_ARN,
        body=json.dumps(payload),
        contentType="application/json",
        accept="application/json",
    )

		

The invoke_model call uses the SageMaker endpoint ARN deployed through Bedrock Marketplace. Chronos-Bolt returns predictions organized by quantile levels, by default, the 10th, 50th (median), and 90th percentiles. This gives us not just a single forecast line, but a confidence band: the 80% interval between the 10th and 90th percentiles tells us how uncertain the model is about its predictions.

The Bedrock runtime client is configured with generous timeouts (120s read, 30s connect) and automatic retries, since inference on time series data can take a moment depending on the history length:

			
def _get_bedrock_runtime_client():
    return boto3.client(
        "bedrock-runtime",
        region_name=AWS_REGION,
        config=Config(
            read_timeout=120,
            connect_timeout=30,
            retries={"max_attempts": 3},
        ),
    )

		

The agent

Wiring everything together is straightforward. We create a BedrockModel pointing to Claude and pass our forecast tool to the Agent:

			
from strands import Agent
from strands.models.bedrock import BedrockModel
from settings import AWS_REGION, Models
from forecast import forecast_time_series
SYSTEM_PROMPT = """You are a time series forecasting assistant powered by Amazon Chronos-Bolt.
You help users predict future values from historical numerical data. When a user provides
time series data or describes a scenario, use the forecast_time_series tool to generate
predictions.
When presenting results:
- Show the median forecast (quantile 0.5) as the main prediction
- Explain the confidence band (quantiles 0.1 and 0.9) as the uncertainty range
- Summarize trends in plain language
"""
def create_agent() -> Agent:
    bedrock_model = BedrockModel(
        model_id=Models.CLAUDE_SONNET,
        region_name=AWS_REGION,
    )
    return Agent(
        model=bedrock_model,
        system_prompt=SYSTEM_PROMPT,
        tools=[forecast_time_series],
    )

		

The system prompt is important here. It tells Claude that it has forecasting capabilities and how to present the results. Without it, the agent would still call the tool correctly (thanks to the Annotated descriptions), but it might not explain the confidence bands or summarize trends as clearly.

Running it

The CLI entry point (cli.py) registers commands and wires everything together. The forecast command generates synthetic hourly data (a sine wave with noise) by default and asks the agent to forecast. You can also pass a custom prompt.

The entry point is minimal:

			
import click
from commands.forecast import run as forecast
@click.group()
def cli():
    pass
cli.add_command(cmd=forecast, name="forecast")
if __name__ == "__main__":
    cli()

		

The actual command lives in commands/forecast.py:

			
@click.command()
@click.option("--prompt", "-p", default=None, help="Custom prompt for the agent.")
def run(prompt: str | None):
    agent = create_agent()
    if prompt is None:
        values = generate_sample_data(num_points=100)
        values_str = ", ".join(f"{v:.2f}" for v in values)
        prompt = (
            f"I have the following hourly sensor readings from the last 100 hours:\n"
            f"[{values_str}]\n\n"
            f"Please forecast the next 24 hours and explain the predicted trend."
        )
    response = agent(prompt)
    click.echo(response)

		

The sine wave is a good choice for a demo because it has a clear periodic pattern that Chronos-Bolt should capture well. With 100 hours of history (about 4 full cycles of a 24-hour pattern), the model has enough data to identify the periodicity and project it forward.

Example

			
(venv) ➜  src python cli.py forecast
2026-02-27 14:11:16,471 - INFO - Found credentials in shared credentials file: ~/.aws/credentials
2026-02-27 14:11:16,506 - INFO - Creating Strands MetricsClient
Sure! Let me run the forecast on your 100-hour sensor readings right away.
Tool #1: forecast_time_series
2026-02-27 14:11:22,981 - INFO - Starting forecast: history=100, prediction_length=24
2026-02-27 14:11:22,981 - INFO - Invoking Chronos-Bolt: history_length=100, prediction_length=24, quantiles=[0.1, 0.5, 0.9]
2026-02-27 14:11:22,994 - INFO - Found credentials in shared credentials file: ~/.aws/credentials
2026-02-27 14:11:23,697 - INFO - Forecast complete: 24 steps, 4 quantiles
Here are the results of your 24-hour forecast:

		

Script response

24-Hour Sensor Forecast

Hour	Low (10th %)	Median (50th %)	High (90th %)
+1	~-0.29	~0.85	~0.99
+2	~0.29	~0.95	~0.99
+3	~0.49	~0.99	~0.99
+4	~0.47	~0.83	~0.99
+5	~0.24	~0.72	~0.96
+6	~0.09	~0.53	~0.83
+7	~-0.25	~0.19	~0.59
+8	~-0.60	~-0.22	~0.25
+9	~-0.91	~-0.51	~-0.04
+10	~-0.99	~-0.76	~-0.30
+11	~-0.99	~-0.87	~-0.46
+12	~-0.99	~-0.93	~-0.56
+13	~-0.99	~-0.96	~-0.57
+14	~-0.99	~-0.93	~-0.52
+15	~-0.99	~-0.87	~-0.44
+16	~-0.99	~-0.79	~-0.35
+17	~-0.96	~-0.63	~-0.16
+18	~-0.86	~-0.46	~0.02
+19	~-0.66	~-0.24	~0.21
+20	~-0.44	~-0.01	~0.42
+21	~-0.21	~0.22	~0.60
+22	~-0.04	~0.45	~0.78
+23	~0.16	~0.66	~0.90
+24	~0.31	~0.80	~0.96

Trend Explanation

Your sensor data clearly exhibits a sinusoidal/oscillating pattern — cycling between peaks near +1.0 and troughs near -1.0 roughly every ~24 hours. The model has picked this up beautifully! Here’s what to expect:

Hours +1 to +4 — Continued Peak : The signal is currently rising (ending at ~0.67–0.70), and the model predicts it will crest near ~+0.85 to +0.99 in the first few hours — completing the current positive phase.
Hours +5 to +13 — Descent & Trough : The signal then descends sharply, hitting a trough around hours +12 to +14 with a median near -0.93 to -0.96 — matching the negative peaks seen in the historical data.
Hours +14 to +24 — Recovery : After bottoming out, the signal climbs back up, reaching approximately +0.80 by hour +24, setting up the next positive cycle.

Confidence Band

The 80% confidence interval (low–high columns) is relatively tight, reflecting the model’s high confidence in the periodic nature of this signal. The widest uncertainty occurs around the transition zones (hours +7–+9 and +17–+19), which is typical for oscillating signals near the zero-crossing points.

In short: your sensor is behaving like a clean oscillating signal with an ~24-hour period, and the next full cycle looks very consistent with historical behavior.Here are the results of your 24-hour forecast:

24-Hour Sensor Forecast

Hour	Low (10th %)	Median (50th %)	High (90th %)
+1	~-0.29	~0.85	~0.99
+2	~0.29	~0.95	~0.99
+3	~0.49	~0.99	~0.99
+4	~0.47	~0.83	~0.99
+5	~0.24	~0.72	~0.96
+6	~0.09	~0.53	~0.83
+7	~-0.25	~0.19	~0.59
+8	~-0.60	~-0.22	~0.25
+9	~-0.91	~-0.51	~-0.04
+10	~-0.99	~-0.76	~-0.30
+11	~-0.99	~-0.87	~-0.46
+12	~-0.99	~-0.93	~-0.56
+13	~-0.99	~-0.96	~-0.57
+14	~-0.99	~-0.93	~-0.52
+15	~-0.99	~-0.87	~-0.44
+16	~-0.99	~-0.79	~-0.35
+17	~-0.96	~-0.63	~-0.16
+18	~-0.86	~-0.46	~0.02
+19	~-0.66	~-0.24	~0.21
+20	~-0.44	~-0.01	~0.42
+21	~-0.21	~0.22	~0.60
+22	~-0.04	~0.45	~0.78
+23	~0.16	~0.66	~0.90
+24	~0.31	~0.80	~0.96

Trend Explanation

Hours +1 to +4 — Continued Peak : The signal is currently rising (ending at ~0.67–0.70), and the model predicts it will crest near ~+0.85 to +0.99 in the first few hours — completing the current positive phase.
Hours +5 to +13 — Descent & Trough : The signal then descends sharply, hitting a trough around hours +12 to +14 with a median near -0.93 to -0.96 — matching the negative peaks seen in the historical data.
Hours +14 to +24 — Recovery : After bottoming out, the signal climbs back up, reaching approximately +0.80 by hour +24, setting up the next positive cycle.

Confidence Band

In short: your sensor is behaving like a clean oscillating signal with an ~24-hour period, and the next full cycle looks very consistent with historical behavior.

And that’s all! Full code in my GitHub account.

Transforming Raw Spreadsheets into Professional Excel Reports with AI Agents and Python

Gonzalo Ayuso — Mon, 09 Feb 2026 13:11:01 +0000

We all deal with spreadsheets. They’re everywhere, financial reports, sales data, operational metrics. But raw data in a flat table is just that: raw data. To extract insights, you need dashboards, charts, KPIs, conditional formatting, and executive summaries. Doing this manually is tedious. What if an AI agent could take any raw .xlsx file and transform it into a professional, multi-sheet workbook with formulas, charts, and insights, automatically?

That’s exactly what this project does. The idea is simple: you give it a spreadsheet, and an AI agent running Python inside a AWS sandbox analyzes the data, builds a Dashboard with KPI formulas, formats the source data, generates an executive summary with real insights, and creates analysis sheets with charts, all using Excel formulas, never hardcoded values.

The two-agent pattern

The core of the system is a two-agent architecture. An outer orchestrator agent (Claude Sonnet) manages the workflow, while an inner agent (Claude Opus) does the actual Excel work inside an AWS Bedrock Code Interpreter sandbox. This separation keeps the orchestration clean and lets the inner agent focus entirely on writing Python code with openpyxl.

The CLI entry point uses Click. When you run the command, it creates the orchestrator agent with the xlsx_enhancer tool:

			
@click.command()
@click.argument("input_file", type=click.Path(exists=True))
@click.argument("output_file", type=click.Path(), required=False)
def run(input_file: str, output_file: str | None):
    if not output_file:
        p = Path(input_file)
        output_file = str(p.parent / f"enhanced_{p.name}")
    agent = create_agent(
        system_prompt=ORCHESTRATOR_PROMPT,
        tools=[xlsx_enhancer],
        hooks=[ToolProgressHook()],
    )
    response = agent(
        f"Process the Excel file at {input_file} and save the enhanced version to {output_file}"
    )
    click.echo(f"Done: {str(response)}")

		

The agent factory wraps the Strands SDK configuration, model selection, retry logic, sliding window conversation management:

			
def create_agent(
    system_prompt: str,
    model: str = Models.CLAUDE_45,
    tools: Optional[List[Any]] = None,
    hooks: Optional[List[HookProvider]] = None,
    temperature: float = 0.3,
    read_timeout: int = 300,
    connect_timeout: int = 60,
    max_attempts: int = 10,
    maximum_messages_to_keep: int = 30,
    should_truncate_results: bool = True,
    callback_handler: Any = None,
) -> Agent:
    bedrock_model = create_bedrock_model(
        model=model,
        temperature=temperature,
        read_timeout=read_timeout,
        connect_timeout=connect_timeout,
        max_attempts=max_attempts,
    )
    return Agent(
        system_prompt=system_prompt,
        model=bedrock_model,
        conversation_manager=SlidingWindowConversationManager(
            window_size=maximum_messages_to_keep,
            should_truncate_results=should_truncate_results,
        ),
        tools=tools,
        hooks=hooks,
        callback_handler=callback_handler,
    )

		

The xlsx_enhancer tool

This is the centerpiece. It’s a Strands @tool that orchestrates a 4-step pipeline: upload the file to the sandbox, run the inner agent, verify the output, and download the result from the sandbox.

			
@tool
def xlsx_enhancer(input_file: str, output_file: str, instructions: str = "") -> dict:
    """Enhance an Excel file with professional formatting, dashboards, charts, and analysis sheets."""
    input_path = Path(input_file)
    output_path = Path(output_file)
    if not input_path.exists():
        return XlsxResult(success=False, error=f"Input file not found: {input_file}").model_dump()
    if input_path.suffix.lower() != ".xlsx":
        return XlsxResult(success=False, error=f"Input file must be .xlsx, got: {input_path.suffix}").model_dump()
    user_prompt = USER_PROMPT
    if instructions.strip():
        user_prompt = f"{USER_PROMPT}\n\n## Additional Instructions\n{instructions}"
    try:
        code_tool = AgentCoreCodeInterpreter(region=AWS_REGION)
        sandbox = SandboxIO(code_tool)
        # 1. Upload
        sandbox.upload(input_path, SANDBOX_INPUT)
        # 2. Run the inner XLSX agent
        agent = create_agent(
            system_prompt=SYSTEM_PROMPT,
            model=Models.CLAUDE_46_OPUS,
            tools=[code_tool.code_interpreter],
        )
        response = agent(user_prompt)
        # 3. Verify output exists in sandbox
        if not sandbox.verify_exists(SANDBOX_OUTPUT):
            return XlsxResult(
                success=False,
                error=f"The XLSX agent did not produce '{SANDBOX_OUTPUT}'",
            ).model_dump()
        # 4. Download
        output_path.parent.mkdir(parents=True, exist_ok=True)
        sandbox.download(SANDBOX_OUTPUT, output_path)
        return XlsxResult(success=True, output_path=str(output_path)).model_dump()
    except SandboxIOError as e:
        return XlsxResult(success=False, error=f"Sandbox I/O failed: {e}").model_dump()

		

The inner agent receives two carefully crafted prompts. The system prompt enforces hard rules about Excel integrity, formulas instead of hardcoded values, sheet name constraints, error handling. The user prompt defines the exact structure: Dashboard with KPI formulas, formatted Data sheet, executive Summary with LLM-generated insights, and Analysis sheets with charts.

The formula-first philosophy

One of the most important design decisions is that the agent never hardcodes computed values in cells. Every number in the output workbook comes from an Excel formula:

			
# FORBIDDEN - Computing in Python
total = df['Sales'].sum()
sheet['B10'] = total  # Hardcodes a value
# REQUIRED - Excel formulas
sheet['B10'] = '=SUM(Data!D:D)'
sheet['C10'] = '=SUMIF(Data!A:A,"Category",Data!B:B)'
sheet['D10'] = '=IFERROR(AVERAGEIF(Data!A:A,A10,Data!D:D),0)'

		

This means the resulting Excel file is alive, change a value in the Data sheet and every KPI, every analysis table, every chart updates automatically. The IFERROR wrapping prevents #DIV/0! errors that would otherwise break AVERAGEIF formulas when a category has no data.

Handling binary files in the sandbox

The AWS Bedrock Code Interpreter sandbox runs Python in an isolated environment. Uploading the source file is straightforward, the bedrock client handles binary blobs natively. But downloading the result is trickier: the download_file method decodes everything as UTF-8, which corrupts binary xlsx files.

The solution is to base64-encode the file inside the sandbox and extract the text from the stream:

			
class SandboxIO:
    def __init__(self, code_tool: AgentCoreCodeInterpreter):
        self._code_tool = code_tool
    def _get_client(self):
        session_name, error = self._code_tool._ensure_session(None)
        if error:
            raise SandboxIOError(f"Failed to ensure session: {error}")
        session_info = self._code_tool._sessions.get(session_name)
        return session_info.client
    def upload(self, local_path: Path, sandbox_name: str = "input.xlsx") -> None:
        file_bytes = local_path.read_bytes()
        client = self._get_client()
        client.upload_file(path=sandbox_name, content=file_bytes)
    def download(self, sandbox_name: str, local_path: Path) -> None:
        client = self._get_client()
        result = client.execute_code(
            "import base64, os\n"
            f"p = '{sandbox_name}'\n"
            "data = open(p, 'rb').read()\n"
            "print(base64.b64encode(data).decode())\n"
        )
        b64_text = _extract_stream_text(result)
        file_bytes = base64.b64decode(b64_text.strip())
        if not file_bytes.startswith(b"PK\x03\x04"):
            raise SandboxIOError("Downloaded file is not a valid xlsx")
        local_path.write_bytes(file_bytes)

		

The PK\x03\x04 check validates the ZIP magic bytes — every xlsx file is a ZIP archive internally.

The original xlsx file

This is the original file we feed into the agent. It’s a flat table with rows and columns. No formatting, no formulas, just bored raw data.

What the agent produces

Given a raw financial spreadsheet, the agent generates a multi-sheet workbook:

Dashboard: KPI cards with formulas (=SUM(Data!D:D), =COUNT(Data!A:A)), color-coded metrics, and a hyperlinked index to all sheets

Data: The original data with dark blue headers, alternating row colors, auto-filters, data bars on numeric columns, and frozen panes

Summary: An executive summary written by the LLM, key findings, concentration risks, trends, anomalies, and actionable recommendations

Analysis sheets: One per categorical column, each with a SUMIF/COUNTIF/AVERAGEIF table and a bar chart

The agent also detects the language of the input data and uses the same language for all generated content, sheet names, titles, labels, and the executive summary.

Monitoring tool execution

A simple hook tracks how long each tool execution takes. It can be extended to integrate with our application and provide real-time feedback to users about the agent’s progress:

			
class ToolProgressHook(HookProvider):
    def __init__(self) -> None:
        self._start_time: float = 0
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeToolCallEvent, self.on_tool_start)
        registry.add_callback(AfterToolCallEvent, self.on_tool_end)
    def on_tool_start(self, event: BeforeToolCallEvent) -> None:
        self._start_time = time.time()
        tool_name = event.tool_use.get("name", "unknown")
        logger.info("Tool started: %s", tool_name)
    def on_tool_end(self, event: AfterToolCallEvent) -> None:
        elapsed = time.time() - self._start_time
        tool_name = event.tool_use.get("name", "unknown")
        logger.info("Tool finished: %s (%.1fs)", tool_name, elapsed)

		

And that’s all. With tools like Strands Agents and AWS Bedrock’s Code Interpreter, we can build AI agents that go beyond text generation, they produce real, functional artifacts. A raw spreadsheet goes in, a professional report comes out. No templates, no manual formatting, just an agent that understands data and knows how to present it.

Full code in my github account.