Turn your OpenAPI in MCP Server in 5 minutes!

This project showcases how to automatically create any MCP server just from the json definition of the openAPI.

What We’ll Build

• An MCP-style FastAPI server that reads IBM COS OpenAPI specs
• Automatically converts API operations into callable tools
• Authenticates using IBM Cloud IAM
• Invokes COS API endpoints dynamically
• Handles region-specific endpoints and bucket names

Prerequisites

• Python 3.8+
• Also the open API documentation json file. For this demo we will use the open API json file for IBM Cloud Object Storage

Step 1: Project Setup

Lets create our project and setup the virtual environment.

mkdir cos-mcp-server
cd cos-mcp-server

Initialise the project

uv init

Initialize the virtual environment

uv venv
source venv/bin/activate

Add all necessary dependencies

uv add fastapi uvicorn httpx dotenv

You can also just clone my Git Repo.

Step 2: Create the MCP Server

We will use the fastMCP module to instantiate our MCP server as follows

from fastapi import FastAPI
app = FastAPI()

Next we need a json file that has the Open API definition. For this example, for this demo I will use the Open API definition for IBM Cloud Object Storage API. The definition has paths for different skills such as ListBucket, CreateBucket and so on as follows:

Open API definition for IBM Cloud Object Storage API

For the MCP server will have following goal:

• input: the url for the Open API definition for any API.
• output: all the paths in the Open API definition are converted as tools

For example, in this case we have given the url for the IBM Cloud Object Storage API and we need a fucntion that will fetch the json file from the url. You can also download the json and store it locally and change the path accordingly.

OPENAPI_URL = "https://cloud.ibm.com/apidocs/cos/cos-compatibility.json"
async def fetch_openapi_spec():
    async with httpx.AsyncClient() as client:
        response = await client.get(OPENAPI_URL)
        response.raise_for_status()
        return response.json()

We need a function that can read each of the path item and can create corresponding MCP tool for our server.

def generate_tools_from_openapi(openapi: Dict[str, Any]):
    paths = openapi.get("paths", {})
    for path, methods in paths.items():
        for method, details in methods.items():
            operation_id = details.get("operationId") or f"{method}_{path.replace('/', '_')}"
            summary = details.get("summary", "")

            # Create a basic tool function with a name and HTTP method
            def make_tool(p, m):
                async def tool_func(input_data):
                    region = input_data.get("region", "us-south")
                    headers = input_data.get("headers", {})
                    body = input_data.get("body", None)
                    params = input_data.get("params", None)
                    params = input_data.get("params", {})
                    formatted_path = p
                    for key, value in params.items():
                        formatted_path = formatted_path.replace(f"{{{key}}}", value)
                    url = f"https://s3.{region}.cloud-object-storage.appdomain.cloud{formatted_path}"
                    async with httpx.AsyncClient() as client:
                        req = client.build_request(m.upper(), url, headers=headers, json=body, params=params)
                        res = await client.send(req)
                        return {"status_code": res.status_code, "body": res.text}
                return tool_func

            tool_registry[operation_id] = make_tool(path, method)

All the tools will be added to In-memory

tool_registry = {}

We can see the tools using the \tools endpoint of the MCP server

@app.get("/tools")
async def list_tools():
    return JSONResponse(content={"tools": list(tool_registry.keys())})

Now we need a mechanism in the server so any client can invoke specific tool. Here is the snippet of our invoke function in the MCP tool.

@app.post("/invoke")
async def invoke_tool(call: ToolCallInput):
    tool_name = call.tool_name
    input_data = call.input
    print(input_data)
    if tool_name not in tool_registry:
        return JSONResponse(status_code=404, content={"error": "Tool not found"})

    tool_func = tool_registry[tool_name]
    try:
        result = await tool_func(input_data)
        return JSONResponse(content={"output": result})
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

Now lets try to have the server up! I know the app.on_event is depreciated and we need use the lifespan API. I ll update it later. For today lets go with it!

@app.on_event("startup")
async def startup():
    openapi = await fetch_openapi_spec()
    generate_tools_from_openapi(openapi)
    print(f"Registered tools: {list(tool_registry.keys())}")

Step 3: Test The server

Start the server:

uv run main.py

Call the /tools endpoint:

curl http://localhost:8000/tools

The output will look like as below, as you can see it has created tools for each path in the OpenAI definition with no additional coding and what so ever.

{"tools":["ListBuckets","CreateBucket","DeleteBucket","HeadBucket","ListObjects","ListObjectsV2","PutObject","GetObject","HeadObject","DeleteObject","CopyObject","InitiateMultipartUpload","CompleteMultipartUpload","ListParts","AbortMultipartUpload","PutBucketLifecycleConfiguration","GetBucketLifecycleConfiguration","DeleteBucketLifecycle","PutBucketReplication","GetBucketReplicationConfiguration","DeleteBucketReplication","RestoreObject","PutBucketProtectionConfiguration","PutBucketWebsite","GetBucketWebsite","DeleteBucketWebsite","PutBucketCors","GetBucketCors","DeleteBucketCors","PutObjectTagging","GetObjectTagging","DeleteObjectTagging","DeleteObjects","PutPublicAccessBlock","GetPublicAccessBlock","DeletePublicAccessBlock","GetBucketAcl","PutBucketAcl","GetObjectAcl","PutObjectAcl","ListMultipartUploads","UploadPart","UploadPartCopy"]}

Step 4: Create a Client to test the Tools in MCP Server

In this post, I’ll walk you through how I built a simple MCP client in Python that talks to a backend MCP server I created for IBM Cloud Object Storage (COS). This client:

• Authenticates using an API key
• Retrieves an IAM token
• Uses tool_call messages to interact with the MCP server
• Lists all storage buckets
• Creates new buckets programmatically

Make sure you have

• The MCP server we just built running as stdio
• A Cloud Object Storage Instance created in your IBM Cloud account.- if you need help to setup a object storage on you account, please find details here.
• The Service Instance ID of your Cloud Object Storage Instance — Follow the steps documented here to find it.
• A valid IBM Cloud API Key — If you need help to create IBM Cloud API Key please check out the documentation.

Step 1: Create Your Project Folder

Lets create our project and setup the virtual environment.

mkdir cos-mcp-client
cd cos-mcp-client

Initialise the project

uv init

Initialize the virtual environment

uv venv
source venv/bin/activate

Add all necessary dependencies

uv add requests

You can also just clone my Git Repo.

Step 2: Create .env to Store Secrets

IBM_CLOUD_API_KEY=your_ibm_cloud_api_key
COS_SERVICE_INSTANCE_ID=your_cos_instance_id
COS_REGION=us-south
MCP_URL="http://localhost:8000/invoke"

Step 3: MCP Client Logic in Python

First we need to get IAM token from IBM cloud to authenticate

def get_iam_token():
    iam_url = "https://iam.cloud.ibm.com/identity/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
    }
    data = {
        "apikey": API_KEY,
        "grant_type": "urn:ibm:params:oauth:grant-type:apikey"
    }
    response = requests.post(iam_url, headers=headers, data=data)
    response.raise_for_status()
    return response.json()["access_token"]

Once we get the token we will be able to invoke any tools in the MCP server. We need to create a json message to invoke corresponding tool. In the too_call Json, we need to define the tool name. If you look at the output of \tools endpoint, the name of the tool to get all buckets from the cloud Obejct storage is “ListBuckets”

tool_call = {
        "tool_name": <tool_name>,
        "input": {
            "headers": {
                "Authorization": f"Bearer {token}",
                "ibm-service-instance-id": COS_SERVICE_INSTANCE_ID
            }
        }
    }

So for getting all the buckets the tool_call will look like

tool_call = {
        "tool_name": "ListBuckets",
        "input": {
            "headers": {
                "Authorization": f"Bearer {token}",
                "ibm-service-instance-id": COS_SERVICE_INSTANCE_ID
            }
        }
    }

Create a file called client.py and paste the following code:

def list_buckets(token):
    tool_call = {
        "tool_name": "ListBuckets",
        "input": {
            "headers": {
                "Authorization": f"Bearer {token}",
                "ibm-service-instance-id": COS_SERVICE_INSTANCE_ID
            }
        }
    }
    response = requests.post(MCP_URL, json=tool_call)
    print("List Buckets:", response.json())

Once I run the client client.py using following command

uv run client.py

For ListBuckets the output is

List Buckets: {'output': {'status_code': 200, 'body': '<?xml version="1.0" encoding="UTF-8" standalone="yes"?><ListAllMyBucketsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Owner><ID>05ae5d0fca40</ID><DisplayName>05ae5d0fca40</DisplayName></Owner><Buckets><Bucket><Name>bkt-images</Name><CreationDate>2022-10-10T02:56:44.050Z</CreationDate></Bucket><Bucket><Name>bx-25-03-2024-29-03-2024</Name><CreationDate>2024-03-30T21:38:30.682Z</CreationDate></Bucket><Bucket><Name>bx-7xl</Name><CreationDate>2020-07-11T12:42:36.195Z</CreationDate></Bucket><Bucket><Name>bx-190385272</Name><CreationDate>2023-11-17T10:43:08.422Z</CreationDate></Bucket><Bucket><Name>bx-02</Name><CreationDate>2020-07-01T16:05:51.424Z</CreationDate></Bucket><Bucket><Name>bx-s8f2bngzf5hqif</Name><CreationDate>2024-06-17T14:55:58.916Z</CreationDate></Bucket><Bucket><Name>bx-41e59bdb07e7</Name><CreationDate>2023-05-31T18:09:33.625Z</CreationDate></Bucket><Bucket><Name>bx-kmx5xkfvtw7nlw</Name><CreationDate>2024-08-11T12:31:11.706Z</CreationDate></Bucket><Bucket><Name>bx-p6ftkz13jbmmjk</Name><CreationDate>2022-09-27T07:58:33.421Z</CreationDate></Bucket><Bucket><Name>bx-0ubdbkw2difgr0</Name><CreationDate>2024-09-18T15:18:16.807Z</CreationDate></Bucket><Bucket><Name>bucket3</Name><CreationDate>2025-04-07T15:37:54.790Z</CreationDate></Bucket></Buckets></ListAllMyBucketsResult>'}}

Similarly for creating a bucket the fucntion will create a tool_call json message and invoke the corresponding tool from the MCP server

def create_bucket(token, bucket_name):
    tool_call = {
        "tool_name": "CreateBucket",
        "input": {
            "region": "us-south",
            "headers": {
                "Authorization": f"Bearer {token}",
                "ibm-service-instance-id": COS_SERVICE_INSTANCE_ID
            },
            "params": {
                "Bucket": bucket_name
            }
        }
    }
    response = requests.post(MCP_URL, json=tool_call)
    print(f"Create Bucket '{bucket_name}':", response.json())

The output of \invoke would be

Create Bucket 'demo-bx-earth': {'output': {'status_code': 200, 'body': ''}}

from fastmcp import FastMCP, Client

mcp = FastMCP.from_fastapi(app) 


async def main():
    openapi = await fetch_openapi_spec()
    generate_tools_from_openapi(openapi)
    print(f"Registered tools: {list(tool_registry.keys())}")
    await mcp.run_sse_async(host="127.0.0.1", port=5050, log_level="debug")

if __name__ == "__main__":
   asyncio.run(main())

So, Now you can use any open API documentation url for json file and create your own MCP server.

Happy MCPing!