OpsfleetAgent Bootcamp

Task 1 · 8 tasks

MCP server on AgentCore Runtime

Your Phase 1 MCP server goes to the cloud behind JWT auth. You own its tools.

20 minMedium
Alice’s ask

The “Tool Server Deployment” challenge

Alice's agent will run in the cloud, so its database tools must too. Take the MCP server from Phase 1 and host it on AgentCore Runtime, reachable only with a valid token.

Platform

What the platform provisions for you

terminal
uv run bootcamp.py up 1
  • Your MCP server code from phase2/app/mcp_server/, packaged and deployed to AgentCore Runtime
  • Inbound JWT auth wired to your Cognito pool from Task 0
  • Your personal copy of the DataStream SQLite database, seeded into your S3 code bucket (DB_BUCKET / DB_KEY)
You

What you do as a developer

  1. Read the deployed server

    phase2/app/mcp_server/mcp_server.py
    DB_BUCKET = os.environ["DB_BUCKET"]
DB_KEY = os.environ.get("DB_KEY", "db/datastream_corp.db")
LOCAL_DB = Path("/tmp/datastream_corp.db")
WRITE_PREFIXES = ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "REPLACE", "TRUNCATE")

mcp = FastMCP("DataStream DB")
s3 = boto3.client("s3")


def local_db() -> Path:
    if not LOCAL_DB.exists():
        s3.download_file(DB_BUCKET, DB_KEY, str(LOCAL_DB))
    return LOCAL_DB


def run_sql(query: str) -> dict:
    with sqlite3.connect(local_db()) as conn:
        conn.row_factory = sqlite3.Row
        cursor = conn.execute(query)
        rows = [dict(row) for row in cursor.fetchall()]
        return {"data": rows, "rows_affected": cursor.rowcount}


@mcp.tool()
def query_db(query: str) -> str:
    """Execute a SQL statement on the DataStream Corp database and return the results as JSON."""
    try:
        result = run_sql(query)
    except sqlite3.Error as error:
        return f"Error executing query: {error}"
    if query.lstrip().upper().startswith(WRITE_PREFIXES):
        s3.upload_file(str(LOCAL_DB), DB_BUCKET, DB_KEY)
    return json.dumps(result, default=str)


if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, stateless_http=True)

    Same tool as Phase 1, two changes. The database file is pulled from your bucket into /tmp on first use and pushed back after every write, so even destructive experiments only touch your data. And the transport is stateless streamable HTTP on port 8000, which AgentCore Runtime expects for the MCP protocol.

  2. Give the model a schema tool

    Challenge

    Models write better SQL when they know the schema. Add a second tool, list_tables, that returns the table names, and tell the model to call it before writing SQL.

    Hint 1

    Every @mcp.tool() function becomes a separate tool, and its docstring is the instruction the model reads. SQLite lists its tables in sqlite_master.

    Hint 2

    Reuse the existing run_sql() helper (don't call query_db from another tool; decorated tools aren't plain functions) and return JSON like query_db does.

    Solution
    phase2/app/mcp_server/mcp_server.py
    @mcp.tool()
def list_tables() -> str:
    """List the tables in the DataStream Corp database. Call this before writing SQL."""
    return json.dumps(run_sql("SELECT name FROM sqlite_master WHERE type = 'table'"))

  3. Deploy and re-test

    terminal
    uv run bootcamp.py deploy
uv run bootcamp.py test --only 1

  4. Experiments

    • Why is stateless_http=True a good fit for a runtime that may scale to many instances? What happens to the /tmp copy if two instances write at once?
    • Sharpen the query_db docstring (SQLite dialect, “read-only preferred”), redeploy, and compare the SQL your agent writes in Task 4.

Check your work

terminal
uv run bootcamp.py test --only 1

Passes when tools/list returns your tools and a SELECT succeeds with a bearer token.

Under the hood

AgentCore Runtime runs your code in isolated, serverless micro-VMs. The CLI builds a Linux/arm64 zip of the folder with its dependencies, uploads it to a code bucket, and the runtime rolls whenever the zip hash changes, so no Docker is needed. The runtime is created with the MCP protocol and a custom JWT authorizer (Cognito discovery URL plus allowed client id), so AgentCore rejects unauthenticated calls before your code runs.