APS を使ったカスタム MCP サーバーの構築

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
const server = new McpServer({
    name: "My APS MCP Server",
    version: "1.0.0",
});

var builder = Host.CreateEmptyApplicationBuilder(settings: null);
builder.Services.AddMcpServer()
    .WithStdioServerTransport()
    .WithToolsFromAssembly();
var app = builder.Build();
await app.RunAsync();

from fastmcp import FastMCP
mcp = FastMCP(
    "My APS MCP Server",
)

app.all("/mcp", async (req, res) => {
    const server = createMcpServer(options);
    const transport = new StreamableHTTPServerTransport({
        sessionIdGenerator: undefined, // No sessions
    });
    res.on("close", () => {
        transport.close();
        server.close();
    });
    await server.connect(transport);
    await transport.handleRequest(req, res, req.body);
});

@mcp.tool()
async def list_hubs(ctx: Context) -> list[dict] | dict:
    token = await _get_valid_token(ctx)
    if not token:
        return {
            "auth_required": True,
            "auth_url": _build_auth_url(ctx.session_id),
            "message": "Open auth_url in a browser to authenticate, then call list_hubs again."
        }
    return await _list_hubs(token)

// .NET example: STDIO transport
builder.Services.AddMcpServer()
    .WithStdioServerTransport()
    .WithToolsFromAssembly();

{
    "mcpServers": {
        "my-server": {
            "command": "dotnet",
            "args": ["run", "--project", "/path/to/mcp-server.csproj"]
        }
    }
}

// JavaScript example: Streamable HTTP with Express
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
const transport = new StreamableHTTPServerTransport({
    sessionIdGenerator: undefined,
});
await server.connect(transport);
await transport.handleRequest(req, res, req.body);

# Python (FastMCP): Streamable HTTP
mcp.run(transport="streamable-http", host="0.0.0.0", port=5000)
# Or via CLI:
# uv run fastmcp run server.py --transport streamable-http --port 5000

async with httpx.AsyncClient() as client:
    response = await client.post(
        APS_TOKEN_URL,
        data={
            "grant_type": "client_credentials",
            "scope": "bucket:read data:read",
        },
        auth=(APS_CLIENT_ID, APS_CLIENT_SECRET),
    )
    data = response.json()

payload = {
    "iss": APS_CLIENT_ID,
    "sub": APS_SSA_ID,
    "aud": APS_TOKEN_URL,
    "exp": now + 300,
    "scope": ["data:read"],
}
assertion = jwt.encode(payload, SSA_PRIVATE_KEY, algorithm="RS256",
                       headers={"kid": SSA_KEY_ID})

// JavaScript equivalent
const assertion = jwt.sign(payload, serviceAccountPrivateKey, {
    algorithm: "RS256",
    header: { alg: "RS256", kid: serviceAccountKeyId },
});

@mcp.custom_route("/callback", methods=["GET"])
async def oauth_callback(request: Request) -> HTMLResponse:
    code = request.query_params.get("code")
    session_id = request.query_params.get("state")
    tokens = await _exchange_code(code)
    _pending_tokens[session_id] = {
        "access_token": tokens["access_token"],
        "refresh_token": tokens.get("refresh_token"),
        "expires_at": time.time() + tokens.get("expires_in", 3600),
    }
    return HTMLResponse("Authentication successful! You can close this window.")

if _token_cache["access_token"] and time.time() < _token_cache["expires_at"] - 60:
    return _token_cache["access_token"]

@mcp.tool()
async def list_buckets() -> list[dict]:
    """List all OSS buckets owned by the configured APS application."""
    token = await _get_access_token()
    return await _list_oss_buckets(token)
@mcp.tool()
async def list_objects(bucket_key: str) -> list[dict]:
    """List objects stored in a specific OSS bucket.
    Args:
        bucket_key: The unique key identifying the OSS bucket.
    """
    token = await _get_access_token()
    return await _list_oss_objects(token, bucket_key)

import z from "zod";
export const getProjectContentsToolFactory = ({}) => ({
    name: "get-project-contents",
    config: {
        title: "Get project contents",
        description: "Retrieves top-level folders in a project or contents of a specified folder.",
        inputSchema: {
            accountId: z.string().nonempty().describe("The ID of the account."),
            projectId: z.string().nonempty().describe("The ID of the project."),
            folderId: z.string().optional().describe("The ID of the folder."),
        },
        annotations: { readOnlyHint: true },
    },
    callback: async ({ accountId, projectId, folderId }) => {
        // ... tool implementation
    },
});

for (const toolFactory of Object.values(tools)) {
    const { name, config, callback } = toolFactory(options);
    registerAppTool(server, name, config, callback);
}

[McpServerToolType]
public static class AECDMTools
{
    [McpServerTool, Description("Get the ACC hubs from the user")]
    public static async Task<string> GetHubs()
    {
        // ... tool implementation
    }
    [McpServerTool, Description("Get the ACC projects from one hub")]
    public static async Task<string> GetProjects(
        [Description("Hub id to query the projects from")] string hubId)
    {
        // ... tool implementation
    }
}

callback: async ({ projectId, designId, region }) => {
    // ... fetch data
    return {
        structuredContent: props.data,  // Full JSON for MCP Apps / UI rendering
        content: [{
            type: "text",
            text: `Found properties containing ${props.data.collection.length} elements`,
        }],
    };
}