Query Amazon Security Lake VPC Flow Logs from AWS DevOps Agent (via a custom MCP server)

16 minute read

Content level: Advanced

AWS DevOps Agent can investigate incidents autonomously, but its built-in Security Lake access can't complete Athena queries against VPC Flow Logs. This article shows how to give the agent reliable, read-only access to that data with a small custom MCP server — deployed from a CloudFormation template.

Introduction

Amazon Security Lake normalizes your VPC Flow Logs into a queryable Apache Iceberg table. AWS DevOps Agent can use that data during an investigation — but when it queries Security Lake through the built-in path, its access session is read-only and can't write Athena's result set to Amazon S3. Queries fail with "Access denied when writing to location."

This article closes that gap with a custom MCP (Model Context Protocol) server:

DevOps Agent → MCP server: the agent calls a single read-only tool, query_vpc_flow, over a SigV4-authenticated Lambda function URL.
MCP server → Security Lake: the Lambda runs the Athena / Lake Formation query under its own execution role — which can write results — and returns structured rows to the agent.

The query runs entirely inside your Security Lake account, so no cross-account data sharing (no Security Lake subscriber, no RAM share) is required. Only the MCP endpoint is exposed, and it is gated by AWS IAM.

This is the MCP server path, not the DevOps Agent Secondary sources path. You do not create a role that trusts aidevops.amazonaws.com; trying to add this as a Secondary source produces an "invalid secondary account" error.

The full flow

AWS DevOps Agent                          Security Lake account
────────────────                          ─────────────────────

 Agent needs VPC flow data
       │
       │ (1) SigV4 (AWS_IAM) POST ──────►  Lambda function URL
       │     request                            │ invokes (sync)
       │                                         ▼
       │                                    MCP Lambda (query_vpc_flow)
       │                                    runs as exec role
       │                                         │
       │                                         ▼
       │                                    Athena query on the
       │                                    vpc_flow Iceberg table
       │                                    (Lake Formation SELECT)
       │                                         │ writes result file
       │                                         ▼
       │                                    S3 results bucket
       │                                         │ Lambda reads results back
       │                                         ▼
       │ (2) HTTP response ◄───────────────  Lambda returns structured
       ▼     (structured rows)               rows in the response body
 Agent uses the rows

The S3 results bucket is Athena's scratch space for the result file. The rows the agent consumes are returned inline in the Lambda's HTTP response to the same function-URL request — a synchronous request/response, not a separate callback.

What you get

A CloudFormation stack (deployed in the Security Lake account) that creates:

Resource	What it does
S3 results bucket	Holds Athena query output (SSE-S3, public access blocked, short lifecycle)
IAM execution role	Lets the Lambda run Athena, read the Glue catalog, write results, and vend Lake Formation credentials
Lake Formation grant	`SELECT` on the vpc_flow table for the execution role
Lambda function	The MCP server — one read-only tool, `query_vpc_flow`, no arbitrary SQL
Lambda function URL	SigV4 (AWS_IAM) HTTPS endpoint the agent calls
Invoke permissions	Allow your Agent Space monitor role to invoke the function

Prerequisites

An AWS DevOps Agent Agent Space already exists in your monitoring account.
Amazon Security Lake is enabled with the VPC Flow Logs (VPC_FLOW / 2.0) source active in your Region.
In the Security Lake account you are a Lake Formation data lake admin with grantable SELECT on the vpc_flow table, plus rights to create IAM/S3/Lambda/CloudFormation resources.
python3, zip, and the AWS CLI v2 installed, with credentials for both accounts.
An S3 bucket in your Region to hold the packaged Lambda code.

⚠️ Every command below changes your AWS account. Test it thoroughly before deploying into production.

Placeholders

Placeholder	Meaning
`<MONITORING_ACCOUNT_ID>`	Account hosting the DevOps Agent Agent Space
`<SECLAKE_ACCOUNT_ID>`	Security Lake / query account (where the stack deploys)
`<REGION>`	Your Security Lake Region, e.g. `us-east-1`
`<SECLAKE_GLUE_DB>`	Glue database, e.g. `amazon_security_lake_glue_db_us_east_1`
`<VPC_FLOW_TABLE>`	e.g. `amazon_security_lake_table_us_east_1_vpc_flow_2_0`
`<MONITOR_ROLE_ARN>`	Your Agent Space service/monitor role ARN (from the Agent Space console settings)
`<MONITOR_ROLE_NAME>`	The name portion of that ARN (drop the `/service-role/` path)
`<CODE_BUCKET>`	S3 bucket in `<REGION>` for the Lambda zip

Step 1: Package the handler

Save the handler (the full app.py is at the end of this article - 'The complete handler') and zip only that file, at the root of the archive:

zip -q devopsagent-securitylake-mcp.zip app.py
unzip -l devopsagent-securitylake-mcp.zip     # must show app.py at the root, no folder prefix
aws s3 cp devopsagent-securitylake-mcp.zip \
  s3://<CODE_BUCKET>/mcp/devopsagent-securitylake-mcp.zip --region <REGION>

The Lambda is dependency-free (it uses boto3, already in the runtime), so there is nothing else to bundle.

If the handler is nested in a folder inside the zip, the Lambda fails with Unable to import module 'app'. Zip from the directory that contains app.py.

Step 2: Deploy the stack

Save the template (at the end of this article - 'The complete CloudFormation template') as devopsagent-securitylake-mcp-infra.yaml, then deploy it in the Security Lake account:

aws cloudformation deploy \
  --template-file devopsagent-securitylake-mcp-infra.yaml \
  --stack-name devopsagent-securitylake-mcp \
  --region <REGION> \
  --capabilities CAPABILITY_NAMED_IAM \
  --parameter-overrides \
    SecurityLakeAccountId=<SECLAKE_ACCOUNT_ID> \
    SecurityLakeDatabaseName=<SECLAKE_GLUE_DB> \
    VpcFlowTableName=<VPC_FLOW_TABLE> \
    McpCodeS3Bucket=<CODE_BUCKET> \
    PrimaryInvokerRoleArn=<MONITOR_ROLE_ARN> \
    SecondaryInvokerRoleArn=<MONITOR_ROLE_ARN(if no secondary account exists)>

Grab the endpoint URL from the outputs — this is generated at deploy time:

aws cloudformation describe-stacks --stack-name devopsagent-securitylake-mcp \
  --region <REGION> --query "Stacks[0].Outputs" --output table

Note the McpEndpointUrl value — you register it in Step 4.

The stack uses fixed resource names (role, function, results bucket). If those names already exist in the account from an earlier deployment, give them unique names before deploying a second stack, or you'll hit "already exists" conflicts.

Step 3: Grant the monitor role permission to invoke

Cross-account invoke needs permission on both sides. The template already added the resource-policy side for the role ARN you passed. Now add the identity-policy side, in the monitoring account:

aws iam put-role-policy --role-name <MONITOR_ROLE_NAME> --policy-name invoke-securitylake-mcp --policy-document '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["lambda:InvokeFunctionUrl","lambda:InvokeFunction"],"Resource":"arn:aws:lambda:<REGION>:<SECLAKE_ACCOUNT_ID>:function:devopsagent-securitylake-mcp"}]}'

Cross-account function-URL invoke requires both lambda:InvokeFunctionUrl and lambda:InvokeFunction. Keep the --policy-document JSON on one line — a multi-line paste can drop later flags.

Step 4: Register the MCP server and associate it

In the DevOps Agent console (monitoring account):

Capability Providers → MCP Server → Register.
Endpoint URL: the McpEndpointUrl from Step 2.
Authorization flow: AWS SigV4. IAM role option: Register without a dedicated role.
AWS Region: <REGION>. Service Name: lambda
Open your Agent Space → Capabilities → MCP Servers → Add, select the server, and allowlist the query_vpc_flow tool.

If association fails with 403 Forbidden at listTools, the monitor role isn't authorized to invoke — recheck Steps 3 and the resource policy for the exact role ARN. If you see 403 Credential should be scoped to correct service: 'lambda', the SigV4 Service Name is wrong; set it to lowercase lambda.

The `query_vpc_flow` tool

The agent calls a single, read-only tool. It validates every input, uses no arbitrary SQL, and caps the row count.

Argument	Purpose
`source_account`	12-digit account whose flow logs to query (required)
`start_time` / `end_time`	`YYYY-MM-DD` or `YYYY-MM-DD HH:MM:SS` (required)
`region`, `src_addr`, `dst_addr`, `src_port`, `dst_port`	Optional filters
`exclude_nodata`	Drop `NODATA` (no-traffic) rows. Default `true`
`full_row`	Return all OCSF columns instead of the curated set. Default `false`
`limit`	Row cap (max 1000)

Know your timestamps: the tool filters on time_dt (the event time), which is not the same as the S3 eventDay delivery-day partition. Query the window that matches when the traffic actually happened. Also, flow records only exist for accounts that had active network interfaces generating traffic.

Security

Invoke = data read: the Lambda runs as the exec role holding the Lake Formation SELECT grant, so grant InvokeFunctionUrl/InvokeFunction only to the specific Agent Space role ARN — never an account ID.
Lock the endpoint: use AuthType=AWS_IAM only (never NONE; enforce org-wide via an SCP on lambda:FunctionUrlAuthType). A raw function URL has no throttling/WAF — front it with CloudFront+WAF or API Gateway if you need that.
Protect data at rest: block public access on the results bucket (the template does) with a short lifecycle; for regulated data use an SSE-KMS CMK and keep the Lake Formation grant scoped to just the vpc_flow table.
Audit: enable CloudTrail + CloudWatch logging and periodically review who can invoke.

Align to your org's posture: the above is a sample sensible baseline — harden further to match your security/compliance needs (SSE-KMS CMKs, reserved concurrency + a dedicated Athena workgroup with a data-scanned cap, private networking via VPC endpoints, and SCP guardrails on invoke).

Cleanup

aws cloudformation delete-stack --stack-name devopsagent-securitylake-mcp --region <REGION>
aws iam delete-role-policy --role-name <MONITOR_ROLE_NAME> --policy-name invoke-securitylake-mcp

Then deregister the MCP server in the DevOps Agent console. (Deleting the stack does not remove query-result objects already written to the bucket if you disabled the lifecycle rule.)

The complete handler

Save as app.py.

"""
MCP server (Streamable HTTP, JSON responses) for querying Amazon Security Lake
VPC flow logs via Athena. Runs as a Lambda behind a function URL (AuthType=AWS_IAM).

Dependency-free: uses only boto3 (bundled in the Lambda Python runtime). Implements
the minimal MCP JSON-RPC surface DevOps Agent needs — initialize, tools/list,
tools/call — and returns application/json (no SSE).

Read-only by design: one tool, parameterized+validated inputs, no arbitrary SQL,
hard LIMIT cap. Data access is enforced by the Lake Formation SELECT grant on the
role this Lambda runs as.
"""

import base64
import json
import os
import re
import time

import boto3

ATHENA_DB = os.environ["SECURITY_LAKE_DATABASE"]
TABLE = os.environ["VPC_FLOW_TABLE"]
RESULTS_LOCATION = os.environ["ATHENA_RESULTS_LOCATION"]
WORKGROUP = os.environ.get("ATHENA_WORKGROUP", "primary")
# OCSF 2.0 field expressions — override via env if the table differs.
TIME_COLUMN = os.environ.get("TIME_COLUMN", "time_dt")          # TIMESTAMP
ACCOUNT_FIELD = os.environ.get("ACCOUNT_FIELD", "accountid")   # top-level partition column
REGION_COLUMN = os.environ.get("REGION_COLUMN", "region")
SRC_IP_EXPR = os.environ.get("SRC_IP_EXPR", "src_endpoint.ip")
DST_IP_EXPR = os.environ.get("DST_IP_EXPR", "dst_endpoint.ip")
STATUS_COLUMN = os.environ.get("STATUS_COLUMN", "status_code")
MAX_LIMIT = int(os.environ.get("MAX_LIMIT", "1000"))
QUERY_TIMEOUT_SECONDS = int(os.environ.get("QUERY_TIMEOUT_SECONDS", "100"))
DEFAULT_PROTOCOL = "2025-06-18"

athena = boto3.client("athena")

SERVER_INFO = {"name": "securitylake-vpcflow", "version": "2.0.0"}

# Lean, high-signal projection (avoids returning huge OCSF structs on every row).
# full_row=true overrides this with SELECT *.
CURATED_COLUMNS = (
    "time_dt, accountid, region, "
    "src_endpoint.ip AS src_ip, src_endpoint.port AS src_port, "
    "dst_endpoint.ip AS dst_ip, dst_endpoint.port AS dst_port, "
    "connection_info.tcp_flags AS tcp_flags, connection_info.direction AS direction, "
    "src_endpoint.interface_uid AS eni, action, disposition, status_code"
)

TOOLS = [
    {
        "name": "query_vpc_flow",
        "description": (
            "Query Amazon Security Lake VPC flow logs (OCSF 2.0) for a source account over a "
            "time range. Read-only. Returns high-signal columns ordered by time (newest first); "
            "NODATA infrastructure-noise rows are excluded by default. Data coverage depends on "
            "Security Lake ingestion for the account/region."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "source_account": {"type": "string", "description": "12-digit AWS account id"},
                "start_time": {"type": "string", "description": "YYYY-MM-DD or 'YYYY-MM-DD HH:MM:SS'"},
                "end_time": {"type": "string", "description": "YYYY-MM-DD or 'YYYY-MM-DD HH:MM:SS'"},
                "region": {"type": "string", "description": "Optional AWS region filter, e.g. us-east-1"},
                "src_addr": {"type": "string", "description": "Optional source IP filter"},
                "dst_addr": {"type": "string", "description": "Optional destination IP filter"},
                "src_port": {"type": "integer", "description": "Optional source port filter"},
                "dst_port": {"type": "integer", "description": "Optional destination port filter"},
                "exclude_nodata": {"type": "boolean", "description": "Drop NODATA (no-traffic) rows. Default true."},
                "full_row": {"type": "boolean", "description": "Return all OCSF columns instead of the curated set. Default false."},
                "limit": {"type": "integer", "description": f"Row cap (max {MAX_LIMIT}). Narrow the window for dense periods."},
            },
            "required": ["source_account", "start_time", "end_time"],
        },
    }
]


# ---------- SQL helpers (validation + escaping; no arbitrary SQL) ----------

def _sql_literal(value):
    if value is None:
        return ""
    if not re.fullmatch(r"[\w:.\-/ ]{0,256}", value):
        raise ValueError("invalid characters in filter value")
    return value.replace("'", "''")


def _timestamp(value, end=False):
    """Accept a date or datetime string; return an Athena TIMESTAMP literal body."""
    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value or ""):
        return value + (" 23:59:59" if end else " 00:00:00")
    if re.fullmatch(r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", value or ""):
        return value
    raise ValueError("time must be 'YYYY-MM-DD' or 'YYYY-MM-DD HH:MM:SS'")


def _run_query(sql):
    started = athena.start_query_execution(
        QueryString=sql,
        QueryExecutionContext={"Database": ATHENA_DB},
        ResultConfiguration={"OutputLocation": RESULTS_LOCATION},
        WorkGroup=WORKGROUP,
    )
    qid = started["QueryExecutionId"]
    deadline = time.time() + QUERY_TIMEOUT_SECONDS
    state = "QUEUED"
    reason = ""
    while time.time() < deadline:
        info = athena.get_query_execution(QueryExecutionId=qid)["QueryExecution"]["Status"]
        state = info["State"]
        reason = info.get("StateChangeReason", "")
        if state in ("SUCCEEDED", "FAILED", "CANCELLED"):
            break
        time.sleep(1.5)
    if state != "SUCCEEDED":
        raise RuntimeError(f"Athena query {state}: {reason or 'timed out'}")

    rows = []
    header = None
    paginator = athena.get_paginator("get_query_results")
    for page in paginator.paginate(QueryExecutionId=qid):
        for record in page["ResultSet"]["Rows"]:
            values = [cell.get("VarCharValue") for cell in record["Data"]]
            if header is None:
                header = values
            else:
                rows.append(dict(zip(header, values)))
    return qid, rows


def _port(value):
    if not re.fullmatch(r"\d{1,5}", str(value)) or int(value) > 65535:
        raise ValueError("port must be an integer 0-65535")
    return int(value)


def tool_query_vpc_flow(args):
    account = args.get("source_account", "")
    if not re.fullmatch(r"\d{12}", account or ""):
        raise ValueError("source_account must be a 12-digit account id")
    start_ts = _timestamp(args.get("start_time", ""), end=False)
    end_ts = _timestamp(args.get("end_time", ""), end=True)
    limit = args.get("limit", 100)
    try:
        limit = max(1, min(int(limit), MAX_LIMIT))
    except (TypeError, ValueError):
        limit = 100

    where = [
        f"{TIME_COLUMN} BETWEEN TIMESTAMP '{start_ts}' AND TIMESTAMP '{end_ts}'",
        f"{ACCOUNT_FIELD} = '{_sql_literal(account)}'",
    ]
    if args.get("exclude_nodata", True):
        where.append(f"{STATUS_COLUMN} <> 'NODATA'")
    if args.get("region"):
        where.append(f"{REGION_COLUMN} = '{_sql_literal(args['region'])}'")
    if args.get("src_addr"):
        where.append(f"{SRC_IP_EXPR} = '{_sql_literal(args['src_addr'])}'")
    if args.get("dst_addr"):
        where.append(f"{DST_IP_EXPR} = '{_sql_literal(args['dst_addr'])}'")
    if args.get("src_port") is not None:
        where.append(f"src_endpoint.port = {_port(args['src_port'])}")
    if args.get("dst_port") is not None:
        where.append(f"dst_endpoint.port = {_port(args['dst_port'])}")

    select = "*" if args.get("full_row") else CURATED_COLUMNS
    sql = (f'SELECT {select} FROM "{ATHENA_DB}"."{TABLE}" WHERE '
           + " AND ".join(where)
           + f" ORDER BY {TIME_COLUMN} DESC LIMIT {limit}")
    qid, rows = _run_query(sql)
    return {"queryExecutionId": qid, "sql": sql, "rowCount": len(rows), "rows": rows}


TOOL_IMPLS = {"query_vpc_flow": tool_query_vpc_flow}


# ---------- MCP JSON-RPC dispatch ----------

def _dispatch(message):
    """Return a JSON-RPC response dict, or None for notifications (no id)."""
    method = message.get("method")
    msg_id = message.get("id")
    params = message.get("params") or {}

    if msg_id is None:  # notification — acknowledge, no body
        return None

    if method == "initialize":
        protocol = params.get("protocolVersion", DEFAULT_PROTOCOL)
        return _ok(msg_id, {
            "protocolVersion": protocol,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": SERVER_INFO,
        })

    if method == "tools/list":
        return _ok(msg_id, {"tools": TOOLS})

    if method == "tools/call":
        name = params.get("name")
        impl = TOOL_IMPLS.get(name)
        if impl is None:
            return _err(msg_id, -32602, f"unknown tool: {name}")
        try:
            payload = impl(params.get("arguments") or {})
        except ValueError as exc:
            return _ok(msg_id, {
                "content": [{"type": "text", "text": f"Invalid arguments: {exc}"}],
                "isError": True,
            })
        except Exception as exc:
            return _ok(msg_id, {
                "content": [{"type": "text", "text": f"Query failed: {exc}"}],
                "isError": True,
            })
        return _ok(msg_id, {
            "content": [{"type": "text", "text": json.dumps(payload, default=str)}],
            "structuredContent": payload,
            "isError": False,
        })

    if method == "ping":
        return _ok(msg_id, {})

    return _err(msg_id, -32601, f"method not found: {method}")


def _ok(msg_id, result):
    return {"jsonrpc": "2.0", "id": msg_id, "result": result}


def _err(msg_id, code, message):
    return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": code, "message": message}}


def _response(status, body):
    return {
        "statusCode": status,
        "headers": {"Content-Type": "application/json"},
        "body": "" if body is None else json.dumps(body),
    }


# ---------- Lambda entrypoint (function URL, payload format 2.0) ----------

def handler(event, context):
    http = (event.get("requestContext") or {}).get("http") or {}
    method = http.get("method", "").upper()
    if method != "POST":
        return _response(405, {"jsonrpc": "2.0", "id": None,
                               "error": {"code": -32600, "message": "only POST is supported"}})

    raw = event.get("body") or ""
    if event.get("isBase64Encoded"):
        raw = base64.b64decode(raw).decode("utf-8")
    try:
        message = json.loads(raw)
    except (ValueError, TypeError):
        return _response(400, {"jsonrpc": "2.0", "id": None,
                               "error": {"code": -32700, "message": "parse error"}})

    if isinstance(message, list):
        responses = [r for r in (_dispatch(m) for m in message) if r is not None]
        return _response(200, responses) if responses else _response(202, None)

    result = _dispatch(message)
    return _response(202, None) if result is None else _response(200, result)

The complete CloudFormation template

Save as devopsagent-securitylake-mcp-infra.yaml.

AWSTemplateFormatVersion: '2010-09-09'
Description: >
  DevOps Agent -> Security Lake VPC Flow MCP server. Provisions the Athena results
  bucket, the MCP Lambda execution role, the Lake Formation SELECT grant on the
  vpc_flow table, the MCP Lambda, its AWS_IAM function URL, and the cross-account
  invoke permissions. Deploy in the Security Lake account with --capabilities CAPABILITY_NAMED_IAM.

Parameters:
  SecurityLakeAccountId:
    Type: String
    Description: The AWS account ID this stack is deployed into (Security Lake / query account).
    AllowedPattern: ^[0-9]{12}$
    ConstraintDescription: Must be a 12-digit AWS account ID.
  SecurityLakeDatabaseName:
    Type: String
    Description: The Glue database that holds the vpc_flow table.
  VpcFlowTableName:
    Type: String
    Default: amazon_security_lake_table_us_east_1_vpc_flow_2_0
  McpCodeS3Bucket:
    Type: String
    Description: S3 bucket holding the packaged MCP Lambda code zip (same region as the stack).
  McpCodeS3Key:
    Type: String
    Default: mcp/devopsagent-securitylake-mcp.zip
  PrimaryInvokerRoleArn:
    Type: String
    Description: The Agent Space monitor role DevOps Agent signs as (override at deploy).
  SecondaryInvokerRoleArn:
    Type: String
    Description: Secondary invoker role (set equal to primary for a single-account build).

Resources:

  ResultsBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub devopsagent-athena-results-${SecurityLakeAccountId}-use1
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        IgnorePublicAcls: true
        BlockPublicPolicy: true
        RestrictPublicBuckets: true
      LifecycleConfiguration:
        Rules:
          - Id: expire-athena-results
            Status: Enabled
            Prefix: athena-results/
            ExpirationInDays: 2

  McpExecRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: DevOpsAgentMcpSecurityLakeExecRole
      Description: Execution role for the Security Lake Athena MCP Lambda
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: securitylake-athena-query
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Sid: AthenaQuery
                Effect: Allow
                Action:
                  - athena:StartQueryExecution
                  - athena:GetQueryExecution
                  - athena:GetQueryResults
                  - athena:StopQueryExecution
                  - athena:GetWorkGroup
                Resource: !Sub arn:aws:athena:${AWS::Region}:${AWS::AccountId}:workgroup/primary
              - Sid: ResultsBucketWrite
                Effect: Allow
                Action:
                  - s3:GetBucketLocation
                  - s3:ListBucket
                  - s3:GetObject
                  - s3:PutObject
                  - s3:AbortMultipartUpload
                  - s3:ListMultipartUploadParts
                Resource:
                  - !GetAtt ResultsBucket.Arn
                  - !Sub ${ResultsBucket.Arn}/*
              - Sid: GlueCatalogRead
                Effect: Allow
                Action:
                  - glue:GetDatabase
                  - glue:GetDatabases
                  - glue:GetTable
                  - glue:GetTables
                  - glue:GetPartition
                  - glue:GetPartitions
                Resource:
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${SecurityLakeDatabaseName}
                  - !Sub arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${SecurityLakeDatabaseName}/${VpcFlowTableName}
              - Sid: LakeFormationCredentialVending
                Effect: Allow
                Action: lakeformation:GetDataAccess
                Resource: '*'

  VpcFlowSelectGrant:
    Type: AWS::LakeFormation::PrincipalPermissions
    DependsOn: McpExecRole
    Properties:
      Principal:
        DataLakePrincipalIdentifier: !GetAtt McpExecRole.Arn
      Resource:
        Table:
          CatalogId: !Ref AWS::AccountId
          DatabaseName: !Ref SecurityLakeDatabaseName
          Name: !Ref VpcFlowTableName
      Permissions:
        - SELECT
        - DESCRIBE
      PermissionsWithGrantOption: []

  McpFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: devopsagent-securitylake-mcp
      Runtime: python3.13
      Handler: app.handler
      Role: !GetAtt McpExecRole.Arn
      Timeout: 120
      MemorySize: 512
      Code:
        S3Bucket: !Ref McpCodeS3Bucket
        S3Key: !Ref McpCodeS3Key
      Environment:
        Variables:
          ATHENA_RESULTS_LOCATION: !Sub s3://${ResultsBucket}/athena-results/
          ATHENA_WORKGROUP: primary
          SECURITY_LAKE_DATABASE: !Ref SecurityLakeDatabaseName
          VPC_FLOW_TABLE: !Ref VpcFlowTableName

  McpFunctionUrl:
    Type: AWS::Lambda::Url
    Properties:
      TargetFunctionArn: !GetAtt McpFunction.Arn
      AuthType: AWS_IAM

  McpUrlInvokePrimary:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref McpFunction
      Action: lambda:InvokeFunctionUrl
      Principal: !Ref PrimaryInvokerRoleArn
      FunctionUrlAuthType: AWS_IAM

  McpUrlInvokeSecondary:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref McpFunction
      Action: lambda:InvokeFunctionUrl
      Principal: !Ref SecondaryInvokerRoleArn
      FunctionUrlAuthType: AWS_IAM

  McpInvokeFunctionPrimary:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref McpFunction
      Action: lambda:InvokeFunction
      Principal: !Ref PrimaryInvokerRoleArn

  McpInvokeFunctionSecondary:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref McpFunction
      Action: lambda:InvokeFunction
      Principal: !Ref SecondaryInvokerRoleArn

Outputs:
  ResultsLocation:
    Value: !Sub s3://${ResultsBucket}/athena-results/
  ExecRoleArn:
    Value: !GetAtt McpExecRole.Arn
  McpEndpointUrl:
    Value: !GetAtt McpFunctionUrl.FunctionUrl

Conclusion

This MCP server gives AWS DevOps Agent reliable, read-only access to your Amazon Security Lake VPC Flow Logs, working around the result-write limitation of the built-in query path. The query runs in your Security Lake account under a scoped execution role, the endpoint is IAM-authenticated, and the agent sees a single safe tool — no arbitrary SQL.

The same pattern can be applied to other Security Lake sources (CloudTrail, Route 53, EKS audit) by pointing the tool at the corresponding table.