48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
"""Read-only MinIO (S3) bridge exposed as MCP tools over SSE."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import boto3
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
BUCKET = os.environ.get("MCP_FS_BUCKET", "agent-workspaces")
|
|
ENDPOINT = os.environ.get("AWS_ENDPOINT_URL", "http://minio.platform-data.svc.cluster.local:9000")
|
|
REGION = os.environ.get("AWS_REGION", "us-east-1")
|
|
HOST = os.environ.get("HOST", "0.0.0.0")
|
|
PORT = int(os.environ.get("PORT", "8080"))
|
|
|
|
_session = boto3.session.Session()
|
|
_client = _session.client(
|
|
"s3",
|
|
endpoint_url=ENDPOINT,
|
|
region_name=REGION,
|
|
aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
|
|
aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
|
|
)
|
|
|
|
mcp = FastMCP("agentic-os-mcp-filesystem", host=HOST, port=PORT)
|
|
|
|
|
|
@mcp.tool()
|
|
def get_object_text(key: str) -> str:
|
|
"""Return the UTF-8 text body for `key` in the configured read-only bucket."""
|
|
obj = _client.get_object(Bucket=BUCKET, Key=key)
|
|
body: bytes = obj["Body"].read()
|
|
return body.decode("utf-8", errors="replace")
|
|
|
|
|
|
@mcp.tool()
|
|
def list_objects(prefix: str = "") -> list[str]:
|
|
"""List object keys under optional prefix (read-only)."""
|
|
keys: list[str] = []
|
|
paginator = _client.get_paginator("list_objects_v2")
|
|
for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
|
|
for item in page.get("Contents", []) or []:
|
|
keys.append(item["Key"])
|
|
return keys
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="sse")
|