Secure Agents with Semantic Kernel, OAuth, and OpenAPI

Acting on your behalf — without exposing everything. Learn how to build trusted, secure agents that work across systems as your digital twin.

A vibrant, futuristic digital scene showing a humanoid AI agent glowing with soft green light, connected via flowing data streams to multiple secure servers, cloud systems, and identity providers. In the background, abstract icons for OAuth, OpenAPI, Semantic Kernel and security shields are subtly integrated into the design. The agent stands confidently and friendly, as if acting on behalf of a user, surrounded by a colorful mesh network of APIs and digital twins. Style is colorful, high-tech, and clean, with gradients and soft lighting, in a modern digital art style.

We’ve all seen the hype posts on LinkedIn and other socials raving about AI agents that can do everything for you. They query your calendar, send your emails, fetch documents, and even talk to APIs. It’s impressive. But there’s something crucial missing from most of those posts: security.

Sure, agents are powerful, especially when they’re tightly integrated into a platform with built-in controls. But what happens when you want your agent to act across systems — your enterprise systems? That’s where things get tricky. Giving an agent access to sensitive tools without opening the floodgates is a challenge most don’t talk about.

This blog post explores a solution where your agent isn’t just some anonymous bot — it’s your digital twin. It uses OAuth to act as you, securely. You log in through a familiar single sign-on provider. From there, the agent uses your token to interact with backend services, all while respecting access boundaries.

We’ll walk through a real-world implementation using:

• A Keycloak-based identity provider (in Docker)
• Two FastAPI-powered backends — one for your app, one simulating an HR system
• An “office manager” agent with elevated access to view other users’ data (securely!)
• A React frontend

If you’re building agent-based systems and care about doing it right, with security built in from the start, this post is for you.

Run Keycloak using Docker.

#

Running Keycloak in Docker is easy for development purposes. Below is a Docker compose config to get you up and running.

version: "3.9"

services:
  keycloak:
    image: quay.io/keycloak/keycloak:24.0.2
    command: start-dev
    ports:
      - "8080:8080"
    environment:
      KEYCLOAK_ADMIN: admin
      KEYCLOAK_ADMIN_PASSWORD: admin
    volumes:
      - keycloak_data:/opt/keycloak/data

volumes:
  keycloak_data:

Writing a blog post about security and still using admin/admin is not smart. Don’t try this in production.

You can access the Keycloak admin tool through the URL: http://localhost:8080. Log in with admin/admin.

Below are the steps to configure a very basic instance of Keycloak:

1. Create a realm: local-dev, I left everything on default.
2. Create a client: fastapi-client , I added valid redirect URLs for the React client and both OpenAPI services.

http://localhost:5173/*

http://localhost:8000/docs/oauth2-redirect

http://localhost:8001/docs/oauth2-redirect

You can use a ‘*’ here, but of course, that is less secure. Note that I am not using HTTPS here, which is also a no-go for production.

3. Add the web origins for the right CORS headers. Add the same hosts here as in the previous step.

4. Create a role office-management

5. Create a group office_managers and assign the role from step 4 to the group.

6. Create two users, jettro and office and put the office user in the office_managers group. Use the switch for email verification. Add a credential, a password. Again, use the switch so people do not have to change the password; the temporary switch must be off.

I got useful information from this blog post:

https://darkaico.medium.com/building-a-secure-authentication-system-with-keycloak-react-and-flask-35aeee04e37a

You should now have a running Keycloak instance with the configuration we need for the demo.

Secured OpenAPI server using FastAPI

#

So many blog posts are available on FastAPI that I do not want to spend too much time on them. However, it is essential to provide the correct information so that the openapi.json file is transparent enough for the Agent to use.

Configure an operation

#

Below is the code for one method.

@app.post("/daysOff", tags=["HR"], responses={401: {"description": "Unauthorized"}},
          summary="Returns number of days off you have left.",
          description="Uses the user in the token to determine the available days off.")
async def days_off(token: str = Security(oauth2_scheme)):
    app_logger.info("Received request to get days off.")
    user_id = verify_token(token)

    if user_id not in days_off_db:
        raise HTTPException(status_code=400, detail="User not found in database.")

    return {"days_off_available": days_off_db[user_id]}

FastAPI provides integration with OAuth providers. Note the parameter oauth2_scheme . This parameter configures the authorisation URL and the token URL. We get back to the verify_token call later in this section.

oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl=KEYCLOAK_AUTH_URL,
    tokenUrl=KEYCLOAK_TOKEN_URL,
)

Create the FastAPI application.

#

As I am creating two different FastAPI services, I created a factory method. This method initialises an app and adds the CORS middleware and the required OAuth configuration to the openapi.json.

def create_app(
        title: str,
        description: str,
        version: str,
        origins: list,
):
    _app = FastAPI(
        title=title,
        description=description,
        version=version,
        swagger_ui_init_oauth={
            "clientId": f"{KEYCLOAK_CLIENT_ID}",
            "usePkceWithAuthorizationCodeGrant": True
        }
    )

    _app.add_middleware(
        CORSMiddleware,
        allow_origins=origins,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    def custom_openapi():
        if _app.openapi_schema:
            return _app.openapi_schema
        openapi_schema = get_openapi(
            title=title,
            version=version,
            description=description,
            routes=_app.routes,
        )
        openapi_schema["components"]["securitySchemes"] = {
            "OAuth2AuthorizationCodeBearer": {
                "type": "oauth2",
                "flows": {
                    "authorizationCode": {
                        "authorizationUrl": f"{KEYCLOAK_BASE_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/auth",
                        "tokenUrl": f"{KEYCLOAK_BASE_URL}/realms/{KEYCLOAK_REALM}/protocol/openid-connect/token",
                        "scopes": {}
                    }
                }
            }
        }
        openapi_schema["security"] = [{"OAuth2AuthorizationCodeBearer": []}]
        _app.openapi_schema = openapi_schema
        return _app.openapi_schema

    _app.openapi = custom_openapi
    return _app

Verify the token

#

To verify the token, we obtain the public key from the Keycloak server. With the public key, the token can be decoded. Now, we can obtain the user’s roles and name. Together with some exception handling, the next code block shows everything you need to verify the token.

It is essential to handle the token with care. Some safety measures are in place, like time-based expiration. But you should always use SSL and not provide URL tokens to be cached or logged. Logging the token like I do for debugging is also not advised.

import logging

import requests
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from fastapi import HTTPException
from jose import jwt, JWTError
from jose.utils import base64url_decode
from jwt import ExpiredSignatureError

from .config import KEYCLOAK_JWKS_URL, KEYCLOAK_ISSUER, AUDIENCE, KEYCLOAK_REALM, KEYCLOAK_CLIENT_ID

# Keycloak authentication and JWT verification utilities for FastAPI.
auth_logger = logging.getLogger("app.auth")

# Cache the JWKS public keys
_jwks = None
def get_jwks():
    global _jwks
    if _jwks is None:
        response = requests.get(KEYCLOAK_JWKS_URL)
        if response.status_code != 200:
            raise Exception("Failed to get JWKS")
        _jwks = response.json()
    return _jwks

def construct_rsa_public_key(jwk: dict):
    n_bytes = base64url_decode(jwk["n"].encode("utf-8"))
    e_bytes = base64url_decode(jwk["e"].encode("utf-8"))
    n = int.from_bytes(n_bytes, byteorder="big")
    e = int.from_bytes(e_bytes, byteorder="big")
    public_key = rsa.RSAPublicNumbers(e, n).public_key(default_backend())
    return public_key

def verify_token(token: str, requested_role: str = None) -> str:
    try:
        auth_logger.debug(f"Verifying token: {token}")

        # Step 1: Fetch JWKS
        jwks = requests.get(KEYCLOAK_JWKS_URL).json()
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")

        # Step 2: Find matching key
        public_key = None
        for key in jwks["keys"]:
            if key["kid"] == kid:
                public_key = construct_rsa_public_key(key)
                break

        if public_key is None:
            auth_logger.error("Public key not found for the given token.")
            raise HTTPException(status_code=401, detail="Public key not found")

        # Step 3: Decode token
        auth_logger.debug(f"Public key found: {public_key}")
        payload = jwt.decode(
            token,
            key=public_key,
            algorithms=["RS256"],
            audience=AUDIENCE,
            issuer=KEYCLOAK_ISSUER
        )
        # Step 4: Check roles if requested
        if requested_role:
            auth_logger.debug(f"Requested role: {requested_role}")
            roles = payload.get("resource_access", {}).get(KEYCLOAK_CLIENT_ID, {}).get("roles", [])
            if requested_role not in roles:
                auth_logger.error(f"User does not have the required role: {requested_role}")
                raise HTTPException(status_code=403, detail="Insufficient permissions")

        return payload.get("preferred_username", payload.get("sub"))

    except ExpiredSignatureError:
        auth_logger.info("Token expired")
        raise HTTPException(status_code=401, detail="Token expired")
    except JWTError as e:
        auth_logger.error(f"JWT error: {str(e)}")
        raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
    except Exception as e:
        # Print everything in the error, including the stack trace
        import traceback
        traceback.print_exc()

        raise HTTPException(status_code=401, detail=f"Unexpected error: {str(e)}")

Now we are ready to test authentication and authorisation using the OpenAPI UI. Browse to the URL http://localhost:8001/docs#/. Note the Authorize button in the top right corner. If the lock is open, you are not authenticated. Using one of the operations /daysOff or /daysOffFor results in a 401. If you click the Authorize button, you are redirected to Keycloak to log in. If you get an error about a redirect URL, check if you use localhost and 127.0.0.1, these are not the same.

Note the open lock in the top right corner button.

Use the client to autorize

The login form for users to login with Keycloak

Note there is a Logout button now, you can close this modal and continue.

Now we can try the operation, use a tryout, and execute it. You should have a response like this.

Note I still have 10 days off

Create the Semantic Kernel Agent

#

The second OpenAPI server acts as the backend for the React application. This service is also secured through OAuth. The operation /query accepts a query or question from the user and uses the agent to reply.

Create the endpoint

#

@app.post("/query", tags=["Agent"], responses={401: {"description": "Unauthorized"}},
          summary="Ask the secure agent",
          description="Send a natural language query and receive an answer")
async def query_agent(request: QueryRequest, token: str = Security(oauth2_scheme)):
    user_id = verify_token(token)

    history = get_user_session(user_id)
    history.add_user_message(request.query)

Note initialising the history. Different users can use the Agent. Therefore, the history is stored using the user name. The function get_user_session obtains the user’s history.

from semantic_kernel.contents import ChatHistory

user_sessions = {}
user_sessions_lock = Lock()

def get_user_session(user_id: str):
    with user_sessions_lock:
        if user_id not in user_sessions:
            history = ChatHistory()
            user_sessions[user_id] = (history)
        return user_sessions[user_id]

Initialise the Kernel

#

In the next code block, we initialise the Kernel and add OpenAIChatCompletion. We use OpenAI as the agent’s LLM. The Semantic Kernel provides default prompts and a strategy for selecting the best tool when multiple tools are available.

kernel = Kernel()
chat_completion = OpenAIChatCompletion(
    api_key=os.getenv("OPENAI_API_KEY"),
    ai_model_id="gpt-4.1",
    service_id="secure-agent",
)
kernel.add_service(chat_completion)
execution_settings = OpenAIChatPromptExecutionSettings()
execution_settings.function_choice_behavior = FunctionChoiceBehavior.Auto()

Add the HR OpenAPI service

#

Semantic Kernel uses plugins to connect with specific extensions. One available plugin is the OpenAPI plugin. This plugin accepts an OpenAPI endpoint to import the operations as tools into the LLM. It is essential to integrate security, as the HR service mandates OAuth. The following code block shows how we add the OpenAPI plugin. Note the auth_callback that adds the required header with the token.

async def add_bearer_token(**kwargs):
    return {"Authorization": f"Bearer {token}"}

kernel.add_plugin_from_openapi(
    plugin_name="hr",
    openapi_document_path="http://localhost:8001/openapi.json",
    execution_settings=OpenAPIFunctionExecutionParameters(
        enable_payload_namespacing=True,
        auth_callback=add_bearer_token,
        server_url_override="http://localhost:8001",
    ),
)

Call the agent

#

We have everything we need to ask the agent for a response.

result = await chat_completion.get_chat_message_content(
    chat_history=history,
    settings=execution_settings,
    kernel=kernel,
)

history.add_message(result)
return {"response": f"{result}"}

Ask the agent how many days off I have left.

Create the React application.

#

Having the OpenAPI UI is nice for testing the application. Integrating it with OAuth is also good. But in the end, we want a better UI. Therefore, we created a React-based application.

Libraries

#

• keycloak-js for the better experience with Keycloak and it OAuth support.
• Axios for streamlined remote calling
• Bootstrap for the easy styling

Initialising Keycloak

#

Ensure the Keycloak initialisation happens only once, or you get an error. The code below initialises Keycloak; you should recognise the values by now.

import Keycloak from 'keycloak-js';

const keycloak = new Keycloak({
  url: 'http://localhost:8080/', // Change if your Keycloak runs elsewhere
  realm: 'local-dev',           // Your Keycloak realm
  clientId: 'fastapi-client',       // Your Keycloak client (should match what FastAPI uses)
});

export default keycloak;

Initialise your React application by wrapping it in the ReactKeycloakProvider .

createRoot(document.getElementById('root')!).render(
    <ReactKeycloakProvider authClient={keycloak}>
        <ErrorBoundary>
            <App/>
        </ErrorBoundary>
    </ReactKeycloakProvider>
)

Configure Axios interceptor

#

All the communication to the backend must use the token as a Bearer in the Authorisation header. The perfect solution for that is the Axios interceptor. The following code block shows how to implement that.

export const useAxiosAuth = () => {
  const { keycloak } = useKeycloak();

  useEffect(() => {
    // Add a request interceptor
    const interceptor = axiosInstance.interceptors.request.use(
      async (config) => {
        if (keycloak?.token) {
          config.headers = config.headers ?? {};
          config.headers['Authorization'] = `Bearer ${keycloak.token}`;
        }
        return config;
      },
      (error) => Promise.reject(error)
    );

    // Cleanup: remove the interceptor when component unmounts or token changes
    return () => {
      axiosInstance.interceptors.request.eject(interceptor);
    };
  }, [keycloak?.token]);
};

Securing routes

#

To make life easier for users, I do not show secured parts of the application. Private routes, in combination with the keycloak functionality, enable hiding routes. They also enable switching the log-in button into a log-out button. Below is the code for the routes to show how that works.

<Routes>
  <Route path="/" element={<Home />} />
  <Route
    path="/dashboard"
    element={
      <PrivateRoute>
        <Dashboard
            user={keycloak.tokenParsed?.preferred_username || 'Unknown'}
            token={keycloak.token || ''}
        />
      </PrivateRoute>
    }
  />
</Routes>

If you want to learn more about the React implementation, please look at the repository for the sample.

The React application showing the response of the agent to my question

One thing to improve

#

One thing I need to improve is working with 401 errors. The sample contains a second endpoint. Someone with the office manager role can also ask about the number of days off available for other people. This works fine. But I get a technical error if I try to use this functionality as a regular user. I want to improve on that.

As jettro I do not have access to days off of other people.

The office user does have access to the office_manager role. This user can check other days off.

Concluding

#

It takes effort, but Semantic Kernel is well-equipped to work with secured endpoints. The integration with OAuth and OpenAPI works well.

You can look at the sample and the code in this repository.

https://github.com/jettro/secure-agent