OpsfleetAgent Bootcamp

Task 2 · 8 tasks

Tools over MCP

Give the agent a key to the company database: a query_db tool served by an MCP server over stdio.

25 minMedium
Alice’s ask

The “Hidden Vault” problem

Alice is impressed, but the real department data is locked in a SQLite database. Prompts can't keep up with 1,200 employees. The agent needs a tool, and you'll expose it the standard way: through the Model Context Protocol.

You

Build it

  1. Expose the database as an MCP tool

    Challenge

    Finish the FastMCP server's one tool, query_db(query: str) -> str, so it runs SQL against DB_PATH, and write its description for the model.

    Work in phase1/starter/mcp_server.py; the starter agent in the next step launches it. Reference: phase1/mcp_server.py.

    Hint 1

    FastMCP turns a typed, documented function into a tool with @mcp.tool(). The docstring becomes the description the model reads, so write it for the model. See FastMCP quickstart.

    Hint 2

    Open the DB with sqlite3.connect, return str(...fetchall()), and return errors as text rather than raising, so the model can read and retry. Run it with mcp.run(transport="stdio").

    Solution
    phase1/mcp_server.py
    mcp = FastMCP("DataStream DB")


@mcp.tool()
def query_db(query: str) -> str:
    """Execute SQL operations on the DataStream Corp SQLite database.

    Supports full CRUD operations: SELECT (read), INSERT (create), UPDATE (modify), DELETE (remove).
    Takes a SQL statement as input and returns the results.
    """
    try:
        with sqlite3.connect(DB_PATH) as conn:
            return str(conn.execute(query).fetchall())
    except Exception as e:
        return f"Error executing query: {e}"


if __name__ == "__main__":
    mcp.run(transport="stdio", show_banner=False)

  2. Connect the agent to it

    Challenge

    Start your server over stdio and give its tools to the agent.

    Work in phase1/starter/t2_mcp_tools.py (look for TODO; an unfinished one prints [starter] TODO …) and run it with uv run bootcamp.py phase1 t2 --starter. The reference solution is phase1/t2_mcp_tools.py; uv run bootcamp.py phase1 t2 runs it.

    Hint 1

    Strands' MCPClient wraps an MCP transport. The starter's make_starter_mcp_client() already builds one that launches your server. See Strands: MCP tools.

    Hint 2

    The client must be open while the agent runs: use it as a context manager and pass mcp_client.list_tools_sync() as tools.

    Solution
    phase1/t2_mcp_tools.py
    with make_mcp_client() as mcp_client:
    agent = Agent(
        name="Data Engineer",
        model=make_model(),
        system_prompt=SYSTEM_PROMPT,
        tools=mcp_client.list_tools_sync(),
    )
    run(agent)

  3. Run it

    terminal · your starter file
    uv run bootcamp.py phase1 t2 --starter "What tables are in the database?"
uv run bootcamp.py phase1 t2 --starter "Describe the employees table"
uv run bootcamp.py phase1 t2 --starter "How many employees are in Engineering?"
    terminal · reference solution
    uv run bootcamp.py phase1 t2 "What tables are in the database?"
uv run bootcamp.py phase1 t2 "Describe the employees table"
uv run bootcamp.py phase1 t2 "How many employees are in Engineering?"

  4. Experiments

    • Watch the tool calls. How many queries does the agent need for a count question? Does it look up the schema first?
    • Mention the table names in the docstring. Does it need fewer calls?
    • This is your copy of the DB, so try a write: “Add a department called Research”. Nothing stops it. Task 5 fixes that.

Check your work

Phase 1 has no automated test: you check it by running the task and looking for the result below.

“What tables are in the database?” returns departments, employees, projects, project_assignments, user_preferences and audit_log (plus SQLite's internal sqlite_sequence). “Describe the employees table” returns columns such as employee_id, first_name, last_name, email, job_title and department_id.

Under the hood

MCP separates who has the tools from who uses them. Here the transport is stdio: make_mcp_client() spawns mcp_server.py with the same Python and talks JSON-RPC over its stdin/stdout. Strands converts each MCP tool schema into a tool spec the model can call. In Phase 2 the same server moves to AgentCore Runtime and the transport becomes streamable HTTP.