Plugin Architecture Across Service Boundaries: An API Contract with Pydantic & FastAPI - Pt. 1

This articles shows how to build a service-level plugin architecture with Pydantic and FastAPI. I will be using an API contract package that plugin providers implement and our platform validates the contract during registration.

Our team operates an internal RAG platform: it crates embeddings from data, stores them in a vector database, and deploys RAG containers that let users interact with their own data.

At first, our platform offered text-embedding models managed by our team. Then other teams wanted to bring their own models - especially for domains such as vision and audio. We needed a way to accept independently deployed services and their models without too much headache for them or us.

The interesting problem was not simply “how much flexibility should our platform offer?” but: how can a plugin service prove that it is compatible with our platform before we let it handle production requests?

I wanted a solution that fit our Python stack, is explicit about request and response shapes (that’s what I call contract), and enforces this across network boundaries. Pydantic is excellent for validating data and is already deeply integrated in FastAPI - but I had not found much practical guidance on using it as the shared API contract for independently deployed services.

In this article, I’ll show the approach we chose: package the Pydantic models for /ready , /catalog, and /encode/ as a shared API-contract library. Plugin authors implement those contracts, while the platform probes and validates each services when it registers. If a response cannot be parsed into the agreed model, registration fails before the plugin becomes available.

Why Use a REST API Contract for a Plugin Architecture?

At first, I was looking for options to provide a strongly typed plugin solution. I also explored other options and weighed them up against each other. After all, choosing the wrong solution at this stage would cost a lot of time and money.

After some research, the following options were left as possible candidates:

1. In-code plugin: Convenient initially, but couples external teams to your repository and release process. Something I want to avoid.
2. Task queue: Flexible, but unnecessary operational complexity for our use case.
3. REST API with a shared contract: Clear ownership and compatibility checks at the boundary

I chose the REST API contract. Either a service fulfills it or it does not. Compared to 1) and 2) the responsibilities are very clear. Option 2) was temping but overkill.

Building the Plugin Architecture

The core idea is to provide a register endpoint at our RAG platform that anyone can call, by providing the URL of the service that should be added to the platform, along with the models. The following shows a simplified version of this. We’ll build up on this diagram as we go.

Once the RAG platform received a register request it probes the URL for three different aspects: 1) Can I embed a test query at POST /encode/. 2) Is the service /ready and (3) does it have a /catalog which tells me more about the models that are hosted.

Relying on the mere API endpoints, however, won’t be enough. At this point, we need a shared data structure on which the RAG Platform can extract the right information from the plugin. That is, for example, how the embedding is represented, e.g. batching, chunks, etc., but also what information the catalogue endpoint serves for the models, e.g. dimension size.

How to Define a Shared API Contract with Pydantic and FastAPI

To solve this, I designed an API Contract, which is based on Pydantic, gets packaged and distributed within our internal Artifactory. This API Contract contains the Pydantic models for the different endpoints. Anyone who creates a plugin service installs this package and annotates the return type of the endpoints with it. The RAG platform itself makes also use of this package by parsing the response of an endpoint directly in the respective model. If it works, the probe is successful. If not, a return value is shown as the response to /register/.

Here, you can see what the different endpoints are expected to accept and to return. Additionally, the package comes with a FastAPI server that is used for testing purposes on my end. You can find out more in Part 2 of this series.

The following shows the FastAPI endpoint with the Pydantic model for the /encode/ endpoint:

class EmbeddingObject(BaseModel):
    """Embedding object with payload and list of floats as embedding."""
    
    embedding: list[float]
    payload: dict 
    
class EncodingInput(BaseModel):
    """Contract for input of the external embedding services."""

    data: list[dict]
    model: str

class EncodingOutput(BaseModel):
    """Contract for encoding output."""
    
    data: list[EmbeddingObject]


@router.post("/encode/", response_model=EncodingOutput, status_code=status.HTTP_201_CREATED)
def post_encode(input: EncodingInput):
    """This endpoint encodes input texts into embeddings."""

    # For demonstration, we return dummy embeddings
    response = encode.EncodingOutput(
        data=[
            encode.EmbeddingObject(
                payload={'page_content': "Test"},
                embedding=[0.1, 0.2, 0.3]
                )
            ]
    )
    return response

In your real world use case, you will have many more fields than you see in this snippet. It’s the same for us. However, for demonstrating the idea and for brevity, it’s sufficient the fields I showed.

You may have noticed that I send the model_id in the body. However, the model is actually the resource I want to run operations against. So, it carries more than just additional information, and arguably, it identifies our resource on plugin side. Ideally, the model would act as a path parameter like this: /models/{model_id}/encode/. Simultaneously, for the catalog endpoint, which is also associated with a model. That tradeoff was made deliberately, because I wanted the API contract as easy as possible for other teams.

Deciding About The Plugin's Responsibility

Defining the endpoints was only one aspect of the contract. We also had to determine who was responsible for what. Our platform manages the data and vector store representation for its built-in text embedding models. However, external plugins may work differently. For instance, a vision plugin might need to load a pre-signed URL independently because the platform does not retain the underlying content or URL (like so in our case).

The same applies to batching: plugin providers know their models' limits best, so the plugin controls how it batches incoming data. These decisions make the contract more detailed, but they keep responsibility with the service that has the necessary domain knowledge.

Quickstart: The Developer Experience

Developers only have to follow three steps to make their service compatible with our RAG platform.

1. uv add plugin-embedding-service-contract
2. Implement the three endpoints (/ready, /encode/, /catalog)
3. Register the service POST /register

That’s really it. And to even speed step 2) up, it’s possible to automatically generate a server stub language agnostic (here for typescript):

brew install openapi-generator
openapi-generator generate -i contract/openapi.yaml -g typescript -o clients/js

How to Handle Authentication, Ownership, and Access

Although I appreciate applications that are as open as possible, I needed to think of some level of user and identity management, for not only using a plugin, but also to execute Create, Update and Delete operations on a plugin.

Since plugged in services execute heavy computational load, it is understandable that teams are free in defining who can access their service. The default is, however, that there isn’t any restriction. Additionally, it’s sensible to differentiate between the subset of people who are able to execute Update and Delete operations. Let’s say the probing was successful. Then, I store a document with the configuration of the plugin service in MongoDB. For a read operation, I check who is allowed to execute the model. For an Update or a Delete, I check for the owner of it. For these kind of verifications, I use FastAPI’s Dependency Injection quite a lot, as it keeps the body of your endpoint clean and focuses on business logic.

Here’s an excerpt:

async def authorize_compute_resource(request: Request, db=Depends(mongodb.get_db)):
    """Authorize access to a compute resource based on a target group.

    The target group is currently decided based on the HTTP method of the request.
    For GET, PUT and DELETE requests, only the owners of the resource are authorized.

    For POST requests, the emails associated with the resource are authorized.
    """
    # 1) extract user token from header
    # 2) extract set of people who can access this resource
    # 3) validate user against set and return the request object if valid
    
# FastAPI endpoint to update a plugin's configuration with dependency injection
@router.put(
    '/compute/{resource_id}',
    status_code=status.HTTP_204_NO_CONTENT,
    dependencies=[Depends(auth_handler.authorize_compute_resource)],
)
async def put_compute_config(request: Request, resource_id: str):
	 # business logic to update a resource

The authorisation code was straightforward compared with deciding the policy. In practice, lifecycle questions — who may use a plugin, who owns it, and who may update or delete it — took more design work than the dependency-injection implementation itself.

How the /register/ Endpoint Validates a Plugin Service

@router.post(
    '/register/',
    status_code=status.HTTP_201_CREATED,
    response_model=compute_model.ComputeServiceRegisterResponse,
    dependencies=[Depends(auth_handler.authorize_compute_creation)],
)
async def post_register_embedding_service(
    register_request: compute_model.ComputeServiceRegister,
    mongo_db=Depends(mongodb.get_db),
):
    """Register an embedding service for a user.

    It probes the service to check if it is reachable and the models are valid.
    """
    api_token = register_request.token.get_secret_value()
    models=register_request.models
    
    probe_result = probe_external_service(register_request.url, models, api_token)
    is_valid = validate_probe(probe_result)
    if not is_valid:
        logger.error('External embedding service registration failed due to invalid API or models.')

    # if yes, store plugin information in a db
    
    return compute_model.ComputeServiceRegisterResponse(...)

I commented out a couple of lines of code as they wouldn't help you understand the idea. HTTPExceptions are too verbose to include here, for example.

At the heart of this approach, I probe the plugin that wants to register itself in probe_external_service. The process is straightforward: A dummy request is made to each endpoint that should be fulfilled by a plugin, and it is validated to ensure that the request and response data adhere to the contract.

def probe_external_service(
    url: str,
    models: dict[str, schemas.CatalogModelInformation | None],
    token: str,
) -> dict:
    """Probe the external embedding service for readiness, catalog, and encode.

    Args:
        url: Base URL of the external embedding service.
        models: model names
        token: API token for authentication.
    """

    # Service-level probes — these affect every model on this service.
    is_ready = probe_external_service_ready(url, token)

    # Per-model probes
    models_result: dict[str, dict[str, bool]] = {}
    for model_name in models:
        is_catalog_valid = probe_external_service_catalog(url, model_name, token)

        probe_data = _get_probe_data(model_name) #dummy values
        is_encoding_valid = probe_external_service_encode(url, model_name, token, probe_data)

        models_result[model_name] = {
            'is_catalog_valid': is_catalog_valid,
            'is_encoding_valid': is_encoding_valid,
        }

    probeness_result = {
        'is_ready': is_ready,
        'is_usage_accessible': is_usage_accessible,
        'models': models_result,
    }

    logger.debug(f'Probe result for external embedding service: {probeness_result}')

    return probeness_result

The probing iterates through each endpoint and the passed models that should be included. Then, it returns a dict of boolean values. Of course, it’d be possible to handle the actual validation in here, but this method is a good example of single responsibility, which is simply to determine if probes are successful. The validation afterwards is a check whether any field is falsy.

Since I do not have control over another service, I decided to probe periodically with Celery Beat. The periodic probing was unseen, but is crucial. You want to detect, once a model is not available, if the plugin is down, or if there’s a drift in agreed values, and set it to false. A service being down for a short period of time happens more often than I thought. As a result, I also added a recovery mode that checks whether inactive models are available again so the plugin becomes active again. In my case, I store the result if valid in a MongoDB, but that’s up to your liking.

Testing Without Container-Based End-to-End Tests

There are many different ways in which we can test this system. I know that, within AWS, engineers would most likely test it with a real Docker container running as a service within CI. While I fully understand this approach, which ensures that everything works because nothing is mocked, it has two disadvantages for small or medium-sized companies in my view: a) Operational costs are high, and b) the most important issue for me is that I want to be faster.

My iteration cycle should be as fast as possible, because adding more external dependencies to your test suite (or tests in general) makes it take longer. At some point, you will feel that the cycle is too long and is holding you back in your development. As I believe testing is crucial, I will explain how I have achieved this in the second part of this series.

The Results // Less Coupling, Little Added Complexity

The majority of the time was spent ensuring that the existing code worked with the new architecture.

I am pretty happy with the results now, as the complexity has not increased much. This is not only due to the API contract, but also to the sense of responsibility.

A significant advantage over the in-code plugin architecture is that we don't accumulate dead code in the sense that a team adds their service but then doesn't use it. With the current approach, they simply shut down their container on a Kubernetes cluster and it becomes unreachable automatically. No code has to be maintained for that. The only effort required is to delete their registration entry in our MongoDB.

Lessons Learned from Building API Contracts

1. Balance abstraction with the details that matter
   It’s always difficult to find the middle ground between abstraction and detail. The more features I wanted to support, the more detailed the contract became.
2. Let the contract evolve through real use
   The data model structure for your API contract is important. However, I don't think it's essential to get it right from the very beginning. You will converge on a good solution based on demand. On the contrary, if you stick to the first version you thought of, it’s a kind of waterfall approach, which won’t work. As you gain practical experience with the contract, you will also rethink the responsibilities of a plugin and a plugin consumer, which will ultimately lead to breaking changes.
3. Iterate on your API resource design
   API design can make your life miserable, but it can also be great. Rushing to a settled version didn’t work for me. I had to iterate and rethink where I want resources to be. I’m sure there are possible improvements, but for now, it is a good start.

What’s next?

A possible evolution of this architecture would be service discovery, so that another plugin in the K8s cluster is automatically detected and added as plugin.

That would be the most convenient approach for plugin providers. This could be done with a label on pod level. While it sounds intriguing from an engineering perspective, I think the decision makes only sense if there were a big number of plugins. So, in that sense, many teams want to be registered with their service.

BUT, it comes at the cost of more responsibility on our stack, and right now, I’d be missing a way to communicate if the API contract is not fulfilled. With the current design, the feedback is clear in form of a response by the /register/ endpoint.

Over to You!

Have you built a plugin system across service boundaries—or taken a different approach to enforcing the API contract? I’d be interested to hear what worked for you. In part two, I’ll show how I test this contract without relying on slow, container-based end-to-end tests.