Mastra AI Framework Guide — Building AI Agents and Workflows

Mastra is an open-source framework for building production-ready AI applications with agents, workflows, tools, memory, and retrieval-augmented generation in a unified architecture.

What You’ll Learn

• Creating AI agents with custom instructions and tool access
• Building multi-step workflows with state management
• Implementing memory, RAG, and tool integration patterns
• Deploying Mastra applications to production

Why Mastra Matters

Building AI applications from scratch requires juggling LLM calls, state management, tool integration, memory, and deployment — all while maintaining reliability. Mastra provides a structured framework that handles these concerns out of the box. DodaTech’s Doda Browser uses Mastra agents for smart tab organization and research workflows. Durga Antivirus Pro prototypes incident response workflows with Mastra’s state machine patterns.

    flowchart LR
    A["Mastra\nFramework"] --> B["Agents\nAI Assistants"]
    A --> C["Workflows\nStep Functions"]
    A --> D["Tools\nCustom Actions"]
    A --> E["Memory\nContext"]
    A --> F["RAG\nRetrieval"]
    B --> G["LLM\nOpenAI/Anthropic"]
    C --> H["Deployment\nProduction"]
    style A fill:#dbeafe,stroke:#2563eb
  

Setting Up Mastra

Install Mastra and initialize a new project.

# pip install mastra
from mastra import Agent, Workflow, Tool
from mastra.llm import OpenAILLM
import os

llm = OpenAILLM(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4")

Expected output: No output — imports are silent. Missing dependencies raise ImportError.

Creating an Agent

Agents are the core building block — AI assistants with system instructions and optional tools.

agent = Agent(
    name="compression_expert",
    instructions="You are a compression algorithm expert at DodaZIP. "
                 "Answer questions about file compression, archiving, and encryption.",
    llm=llm
)

response = agent.run("What compression method should I use for text files?")
print(response)

Expected output:

For text files, Deflate (the algorithm used in ZIP and gzip) is generally the best choice. It combines LZ77 sliding window compression with Huffman coding, achieving good compression ratios with fast performance. For maximum compression, consider LZMA (used in 7z), which is slower but produces smaller outputs.

Adding Tools to Agents

Tools extend what an agent can do. Define tools as Python functions with descriptions.

def get_file_size(filepath: str) -> str:
    import os
    size = os.path.getsize(filepath)
    return f"File size: {size} bytes ({size/1024:.1f} KB)"

tool = Tool(
    name="get_file_size",
    description="Get the size of a file on disk",
    function=get_file_size
)

agent = Agent(
    name="file_analyzer",
    instructions="You help users analyze files. Use tools when needed.",
    llm=llm,
    tools=[tool]
)

response = agent.run("How big is the document.pdf file?")
print(response)

Expected output (varies based on actual file):

Let me check the file size for you.

The file document.pdf is 245,760 bytes (approximately 240 KB). This is a moderate size for a PDF document.

The agent automatically decided to call the get_file_size tool, received the result, and incorporated it into a natural language response.

Building Workflows

Workflows are multi-step processes with state management, similar to state machines or directed acyclic graphs (DAGs).

from mastra import Workflow, Step

class AnalyzeFile(Step):
    def execute(self, state):
        filepath = state["filepath"]
        return {"file_type": filepath.split('.')[-1]}

class ClassifyThreat(Step):
    def execute(self, state):
        file_type = state["file_type"]
        # Simulated classification logic
        if file_type in ["exe", "dll", "scr"]:
            return {"risk_level": "high", "requires_scan": True}
        elif file_type in ["pdf", "docx"]:
            return {"risk_level": "medium", "requires_scan": True}
        else:
            return {"risk_level": "low", "requires_scan": False}

class GenerateReport(Step):
    def execute(self, state):
        return {
            "report": f"File: {state['filepath']}\n"
                      f"Type: {state['file_type']}\n"
                      f"Risk: {state['risk_level']}"
        }

workflow = Workflow(
    name="threat_analysis",
    steps=[AnalyzeFile(), ClassifyThreat(), GenerateReport()]
)

result = workflow.run({"filepath": "downloaded/setup.exe"})
print(result["report"])

Expected output:

File: downloaded/setup.exe
Type: exe
Risk: high

Each step receives and returns state, forming a pipeline. Workflows track execution history, support branching, and can include LLM calls within steps. Durga Antivirus Pro uses this pattern for its automated threat triage pipeline.

Memory and Context

Agents can maintain conversation history across interactions.

agent = Agent(
    name="research_assistant",
    instructions="You help with research. Remember previous context.",
    llm=llm,
    memory=True
)

agent.run("My name is Alex and I'm researching compression algorithms.")
response = agent.run("What was I researching and what's my name?")
print(response)

Expected output:

You're Alex, and you're researching compression algorithms. I remember you asked about file compression methods earlier. How can I help you with your research today?

RAG with Mastra

Mastra supports retrieval-augmented generation with vector stores.

from mastra import Document, VectorStore

# Create documents
doc1 = Document(
    content="DodaZIP supports ZIP, 7z, RAR, and TAR formats with AES-256 encryption.",
    metadata={"source": "dodazip_docs.txt"}
)
doc2 = Document(
    content="Maximum compression is achieved using LZMA2 with solid archiving.",
    metadata={"source": "dodazip_docs.txt"}
)

# Store and retrieve
vector_store = VectorStore()
vector_store.add_documents([doc1, doc2])

agent = Agent(
    name="doc_qa",
    instructions="Answer questions based on the provided documentation.",
    llm=llm,
    rag=vector_store
)

response = agent.run("What encryption does DodaZIP use?")
print(response)

Expected output:

Based on the documentation, DodaZIP supports AES-256 encryption for secure file archiving. It's compatible with ZIP, 7z, RAR, and TAR formats.

Deployment

Mastra applications can be deployed as APIs, serverless functions, or background workers.

from mastra import serve

# Expose agent as an API endpoint
serve(agent, path="/api/assistant", port=8080)
# Now accessible at http://localhost:8080/api/assistant

Expected output: Server starts on port 8080. Send POST requests with {"message": "Hello"} to interact with the agent via HTTP.

Common Errors

1. LLM Configuration Missing

The LLM must be configured with a valid API key. Verify OPENAI_API_KEY (or your provider’s key) is set before creating agents. Missing keys raise AuthenticationError.

2. Tool Function Signature Mismatch

Tool functions must accept and return strings. If your function returns a dict, wrap it in str(). Type hints are optional but recommended for documentation.

3. Workflow Step Ordering

Steps execute in the order they are passed to the Workflow constructor. Ensure each step’s output keys match the next step’s expected input keys.

4. Memory Growth

Conversation memory grows with each interaction. Configure max_memory_tokens or use sliding window memory to prevent context overflow in long-running agents.

5. Vector Store Persistence

By default, vector stores are in-memory. For production, configure a persistent backend (Chroma, Pinecone, Qdrant) to avoid data loss on restart.

6. Workflow Timeouts

Workflows with many steps or slow LLM calls may time out. Set appropriate timeout_seconds on the workflow and implement retry logic for individual steps.

7. Dependency Conflicts

Mastra depends on specific versions of pydantic, httpx, and LLM SDKs. Use a virtual environment and pin versions in requirements.txt.

Practice Questions

1. What are the four core components of Mastra?
2. How do tools differ from workflow steps?
3. What type of state management does Mastra workflows use?
4. How do you add RAG capabilities to a Mastra agent?
5. How do you deploy a Mastra agent as an API?

Answers:

1. Agents (AI assistants), Workflows (multi-step state machines), Tools (custom functions), Memory (conversation history). RAG and LLM integration are built-in extensions.
2. Tools are called dynamically by agents based on conversation context. Workflow steps execute in a fixed order defined by the developer, with explicit state passing between steps.
3. Workflows use a dictionary-based state object that each step reads from and writes to. The state persists across all steps in a single execution but is not shared across different workflow runs.
4. Create a VectorStore, add documents to it, then pass it as the rag parameter when creating an Agent. The agent automatically retrieves relevant documents for each query.
5. Use serve(agent, path="/api/assistant", port=8080) which starts an HTTP server. The agent accepts POST requests and returns responses as JSON.

Challenge: Doda Browser needs a smart bookmark organizer. Build a Mastra workflow that: (1) receives a list of bookmarks, (2) categorizes each using an LLM, (3) groups them into folders, and (4) returns an organized folder structure. Use tools to validate URLs and detect duplicates.

Mini Project: Smart File Assistant Agent

Build an agent that helps users manage and analyze files using tools and memory:

from mastra import Agent, Tool
from mastra.llm import OpenAILLM
import os

llm = OpenAILLM(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4")

def list_files(directory: str = ".") -> str:
    files = os.listdir(directory)
    return "\n".join(files)

def read_file(filepath: str) -> str:
    with open(filepath) as f:
        return f.read()

agent = Agent(
    name="file_assistant",
    instructions="Help users manage files. List directories, read files, "
                 "and answer questions about file contents.",
    llm=llm,
    tools=[
        Tool(name="list_files", description="List files in a directory", function=list_files),
        Tool(name="read_file", description="Read the contents of a file", function=read_file),
    ],
    memory=True
)

print(agent.run("List the files in the current directory"))
print(agent.run("Read the first file and summarize it"))

Try it: Run the agent in a directory with text files. The agent will list files, read them, and provide summaries using the LLM. Extend it with compression analysis tools for DodaZIP integration.

FAQ

Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.