Open source server for AI resources. Built on FastAPI and Pydantic.
from .ai_server import RestAiServer
from .knowledge_base import KnowledgeBase, SemanticSearchResult
from .agent import Agent

aiser/__init__.py
penlight-ai/aiser
from abc import ABC, abstractmethod
import typing

from ..identifiable_entities import IdentifiableEntity
from ..models import ChatMessage


class Agent(IdentifiableEntity, ABC):
    def __init__(self, agent_id: str):
        super().__init__(entity_id=agent_id)

    @abstractmethod
    def reply(self, messages: typing.List[ChatMessage]) -> typing.AsyncGenerator[ChatMessage, None]:
        raise NotImplementedError

aiser/agent/agent.py
penlight-ai/aiser
from .rest_ai_server import RestAiServer

aiser/ai_server/__init__.py
penlight-ai/aiser
from abc import ABC, abstractmethod
from typing import List, Optional

from ..knowledge_base import KnowledgeBase
from ..agent import Agent
from ..config import AiServerConfig, make_ai_server_config
from ..version import __version__


class AiServer(ABC):
    _knowledge_bases: List[KnowledgeBase]
    _agents: List[Agent]
    _port: int

    def __init__(
            self,
            host: str,
            port: int,
            complete_url: Optional[bool],
            knowledge_bases: Optional[List[KnowledgeBase]] = None,
            agents: Optional[List[Agent]] = None,
            config: Optional[AiServerConfig] = None
    ):
        super().__init__()
        self._port = port
        self._host = host
        self._knowledge_bases = [] if knowledge_bases is None else knowledge_bases
        self._agents = [] if agents is None else agents
        self._config: AiServerConfig = config or make_ai_server_config(
            complete_url=complete_url
        )

    def get_aiser_version(self) -> str:
        return __version__

    @abstractmethod
    def run(self):
        raise NotImplementedError

    def _get_list_of_identifiable_entity_ids(self) -> List[str]:
        agent_ids = [
            agent.get_id()
            for agent in self._agents
        ]
        knowledge_base_ids = [
            knowledge_base.get_id()
            for knowledge_base in self._knowledge_bases
        ]
        return agent_ids + knowledge_base_ids

aiser/ai_server/ai_server.py
penlight-ai/aiser
from .asymmetric_jwt_rest_authenticator import AsymmetricJwtRestAuthenticator
from .non_functional_rest_authenticator import NonFunctionalRestAuthenticator
from .rest_authenticator import RestAuthenticator

aiser/ai_server/authentication/__init__.py
penlight-ai/aiser
import typing
import jwt
import httpx
import time
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2

from aiser.ai_server.authentication.rest_authenticator import RestAuthenticator, TokenVerificationCallable
from aiser.utils import base64_to_pem
from aiser.models.dtos import PublicKeyInfo
from aiser.ai_server_consumer import AiServerConsumer


class PublicKeyInfoClient:
    def __init__(self, consumer: AiServerConsumer):
        self._consumer = consumer

    async def fetch_public_key_info(self) -> PublicKeyInfo:
        async with httpx.AsyncClient() as client:
            url = str(self._consumer.publicKeyInfoUrl)
            response = await client.get(url)
            response.raise_for_status()
            public_key_info = PublicKeyInfo(**response.json())
            return public_key_info


class PublicKeyInfoGetter:
    def __init__(
            self,
            public_key_info_client: PublicKeyInfoClient,
            refresh_interval_in_seconds: float = 60,
    ):
        self._public_key_info_client = public_key_info_client
        self._public_key_info: typing.Optional[PublicKeyInfo] = None
        self._last_refresh_timestamp: float = 0
        self._refresh_interval_in_seconds = refresh_interval_in_seconds

    async def get_public_key_info(self) -> PublicKeyInfo:
        if (self._public_key_info is None
                or (time.time() - self._last_refresh_timestamp) > self._refresh_interval_in_seconds):
            self._public_key_info = await self._public_key_info_client.fetch_public_key_info()
            self._last_refresh_timestamp = time.time()
        return self._public_key_info


class AsymmetricJwtRestAuthenticator(RestAuthenticator):
    def __init__(
            self,
            complete_server_url: typing.Optional[str],
            consumer: AiServerConsumer
    ):
        self._complete_server_url = complete_server_url
        self._consumer = consumer

    def _make_authentication_dependency(self, acceptable_subjects: typing.List[str]) -> TokenVerificationCallable:
        public_key_info_client = PublicKeyInfoClient(consumer=self._consumer)
        public_key_info_getter = PublicKeyInfoGetter(public_key_info_client=public_key_info_client)

        auth_scheme = OAuth2()

        async def verify_token(token: str = Depends(auth_scheme)) -> str:
            try:
                token_without_prefix = token.split(" ")[1]
                public_key = (await public_key_info_getter.get_public_key_info()).publicKey
                public_key_pem = base64_to_pem(public_key)
                decoded_jwt = jwt.decode(jwt=token_without_prefix, key=public_key_pem, algorithms=["RS256"], options={
                    "verify_signature": True,
                    "verify_exp": True,
                    "verify_nbf": True,
                    "verify_iat": True,
                    "verify_aud": False,
                })

                if (self._complete_server_url is not None) and (decoded_jwt['aud'] != self._complete_server_url):
                    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
                if decoded_jwt['sub'] not in acceptable_subjects:
                    raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
            except jwt.exceptions.InvalidTokenError as error:
                raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
            return token

        return verify_token

    def get_authentication_dependency(self, acceptable_subjects: typing.List[str]) -> TokenVerificationCallable:
        return self._make_authentication_dependency(acceptable_subjects=acceptable_subjects)

aiser/ai_server/authentication/asymmetric_jwt_rest_authenticator.py
penlight-ai/aiser
import typing
from aiser.ai_server.authentication.rest_authenticator import RestAuthenticator, TokenVerificationCallable


class NonFunctionalRestAuthenticator(RestAuthenticator):
    def get_authentication_dependency(self, acceptable_subjects: typing.List[str]) -> TokenVerificationCallable:
        async def verify_token() -> str:
            return ''

        return verify_token

aiser/ai_server/authentication/non_functional_rest_authenticator.py
penlight-ai/aiser
from abc import ABC, abstractmethod
import typing

TokenVerifyCoroutine = typing.Coroutine[typing.Any, typing.Any, str]
TokenVerificationCallable = typing.Callable[[...], TokenVerifyCoroutine]


class RestAuthenticator(ABC):
    @abstractmethod
    def get_authentication_dependency(self, acceptable_subjects: typing.List[str]) -> TokenVerificationCallable:
        raise NotImplementedError

aiser/ai_server/authentication/rest_authenticator.py
penlight-ai/aiser
from .rest_ai_server import RestAiServer

aiser/ai_server/rest_ai_server/__init__.py
penlight-ai/aiser
import time
import typing

import uvicorn
from fastapi import FastAPI, HTTPException, Depends, APIRouter, Request, status
from fastapi.responses import StreamingResponse
from aiser.ai_server.ai_server import AiServer
from aiser.ai_server.authentication import (
    AsymmetricJwtRestAuthenticator,
    NonFunctionalRestAuthenticator,
    RestAuthenticator
)
from aiser.config.ai_server_config import ServerEnvironment
from aiser.models.dtos import (
    SemanticSearchRequest,
    AgentChatRequest,
    SemanticSearchResultDto,
    SemanticSearchResultResponseDto,
    AgentChatResponse,
    ChatMessageDto,
    VersionInfo
)
from aiser.models import ChatMessage
from aiser.knowledge_base import KnowledgeBase
from aiser.agent import Agent
from aiser.config import AiServerConfig
from aiser.utils import meets_minimum_version


class RestAiServer(AiServer):
    def __init__(
            self,
            complete_url: typing.Optional[str] = None,
            knowledge_bases: typing.Optional[typing.List[KnowledgeBase]] = None,
            agents: typing.Optional[typing.List[Agent]] = None,
            host: str = "127.0.0.1",
            port: int = 5000,
            workers: typing.Optional[int] = None,
            config: typing.Optional[AiServerConfig] = None,
            authenticator: typing.Optional[RestAuthenticator] = None
    ):
        super().__init__(
            complete_url=complete_url,
            knowledge_bases=knowledge_bases,
            agents=agents,
            host=host,
            port=port,
            config=config
        )
        self._workers = workers
        self._authenticator = authenticator or self._determine_authenticator_fallback()

    def _determine_authenticator_fallback(self) -> RestAuthenticator:
        if self._config.server_environment == ServerEnvironment.DEVELOPMENT:
            return NonFunctionalRestAuthenticator()
        return AsymmetricJwtRestAuthenticator(
            complete_server_url=self._config.complete_url,
            consumer=self._config.consumer,
        )

    def get_app(self) -> FastAPI:
        verify_token = self._authenticator.get_authentication_dependency(
            acceptable_subjects=self._get_list_of_identifiable_entity_ids()
        )

        def verify_meets_minimum_version(request: Request):
            min_version = request.headers.get("Min-Aiser-Version")
            if min_version is None:
                return
            if not meets_minimum_version(
                    current_version=self.get_aiser_version(),
                    min_version=min_version
            ):
                error_message = f"Minimum version required: {min_version}. Current version: {self.get_aiser_version()}"
                print(error_message)
                raise HTTPException(
                    status_code=status.HTTP_426_UPGRADE_REQUIRED,
                    detail=error_message
                )

        non_authenticated_router = APIRouter()
        authenticated_router = APIRouter(dependencies=[
            Depends(verify_token),
            Depends(verify_meets_minimum_version)
        ])

        @non_authenticated_router.get("/")
        async def read_root():
            return "ok"

        @authenticated_router.get("/version")
        async def version() -> VersionInfo:
            return VersionInfo(version=self.get_aiser_version())

        @authenticated_router.post("/knowledge-base/{kb_id}/semantic-search")
        async def knowledge_base(
                kb_id: str, request: SemanticSearchRequest,
        ) -> SemanticSearchResultResponseDto:
            for kb in self._knowledge_bases:
                if kb.accepts_id(kb_id):
                    results = kb.perform_semantic_search(
                        query_text=request.text,
                        desired_number_of_results=request.numResults
                    )
                    result_dto = SemanticSearchResultResponseDto(results=[
                        SemanticSearchResultDto(content=result.content, score=result.score)
                        for result in results
                    ])
                    return result_dto
            raise HTTPException(status_code=404, detail="Knowledge base not found")

        async def convert_agent_message_gen_to_streaming_response(
                message_gen: typing.AsyncGenerator[ChatMessage, None]) -> typing.AsyncGenerator[str, None]:
            async for item in message_gen:
                message_dto = ChatMessageDto(textContent=item.text_content)
                yield AgentChatResponse(outputMessage=message_dto).model_dump_json(by_alias=True) + "\n"

        @authenticated_router.post("/agent/{agent_id}/chat")
        async def agent_chat(
                agent_id: str,
                request: AgentChatRequest,
        ) -> StreamingResponse:
            for agent in self._agents:
                if agent.accepts_id(agent_id):
                    messages = [ChatMessage(text_content=messageDto.textContent) for messageDto in request.messages]
                    response_generator = agent.reply(messages=messages)
                    response_generator = convert_agent_message_gen_to_streaming_response(message_gen=response_generator)
                    return StreamingResponse(
                        response_generator,
                        media_type="text/event-stream"
                    )
            raise HTTPException(status_code=404, detail="Agent not found")

        app = FastAPI()
        app.include_router(authenticated_router)
        app.include_router(non_authenticated_router)

        return app

    def run(self):
        uvicorn.run(app=self.get_app(), port=self._port, host=self._host, workers=self._workers or 1)

aiser/ai_server/rest_ai_server/rest_ai_server.py
penlight-ai/aiser
from .ai_server_consumer import AiServerConsumer
from .predefined_consumers import get_default_consumer

aiser/ai_server_consumer/__init__.py
penlight-ai/aiser
from pydantic import BaseModel, AnyHttpUrl


class AiServerConsumer(BaseModel):
    publicKeyInfoUrl: AnyHttpUrl

aiser/ai_server_consumer/ai_server_consumer.py
penlight-ai/aiser
from .ai_server_consumer import AiServerConsumer


class PredefinedConsumers:
    penlight_ai = AiServerConsumer(
        publicKeyInfoUrl="https://api.penlight.ai/public-key"
    )


def get_default_consumer() -> AiServerConsumer:
    return PredefinedConsumers.penlight_ai

aiser/ai_server_consumer/predefined_consumers.py
penlight-ai/aiser
from .ai_server_config import AiServerConfig, make_ai_server_config

aiser/config/__init__.py
penlight-ai/aiser
from .ai_server_env_variables import get_ai_server_env_variables, AiServerEnvVariables
from ..ai_server_consumer import AiServerConsumer, get_default_consumer
import typing
from pydantic import BaseModel


class ServerEnvironment:
    DEVELOPMENT = 'Development'
    PRODUCTION = 'Production'


class AiServerConfig(BaseModel):
    consumer: AiServerConsumer
    complete_url: typing.Optional[str]
    server_environment: str


def make_ai_server_config(
        complete_url: typing.Optional[str],
        consumer: typing.Optional[AiServerConsumer] = None
) -> AiServerConfig:
    config_maker = AiServerConfigMaker(
        complete_url=complete_url,
        env_variables=get_ai_server_env_variables(),
        overriding_consumer=consumer
    )
    return config_maker.make_config()


class AiServerConfigMaker:
    def __init__(
            self,
            complete_url: typing.Optional[str],
            env_variables: AiServerEnvVariables,
            overriding_consumer: typing.Optional[AiServerConsumer] = None,
    ):
        self.complete_url = complete_url
        self.env_variables = env_variables
        self.overriding_consumer = overriding_consumer

    def make_config(self) -> AiServerConfig:
        final_consumer = self._determine_consumer()
        final_server_environment = self._determine_server_environment()

        return AiServerConfig(
            complete_url=self.complete_url,
            consumer=final_consumer,
            server_environment=final_server_environment
        )

    def _determine_consumer(
            self,
    ) -> AiServerConsumer:
        if self.overriding_consumer is not None:
            return self.overriding_consumer
        if self.env_variables.CONSUMER_PUBLIC_KEY_INFO_URL is not None:
            return AiServerConsumer(
                publicKeyInfoUrl=self.env_variables.CONSUMER_PUBLIC_KEY_INFO_URL
            )
        return get_default_consumer()

    def _determine_server_environment(
            self
    ) -> str:
        if self.env_variables.AISER_ENVIRONMENT == ServerEnvironment.DEVELOPMENT:
            return ServerEnvironment.DEVELOPMENT
        return ServerEnvironment.PRODUCTION

aiser/config/ai_server_config.py
penlight-ai/aiser
import os
import typing
from pydantic import BaseModel, AnyHttpUrl


class AiServerEnvVariables(BaseModel):
    CONSUMER_PUBLIC_KEY_INFO_URL: typing.Optional[AnyHttpUrl]
    AISER_ENVIRONMENT: typing.Optional[str]


def get_ai_server_env_variables() -> AiServerEnvVariables:
    return AiServerEnvVariables(
        CONSUMER_PUBLIC_KEY_INFO_URL=os.environ.get("CONSUMER_PUBLIC_KEY_INFO_URL"),
        AISER_ENVIRONMENT=os.environ.get("AISER_ENVIRONMENT")
    )

aiser/config/ai_server_env_variables.py
penlight-ai/aiser
from .load_dotenv_if_available import load_dotenv_if_available

aiser/env_tools/__init__.py
penlight-ai/aiser
from dotenv import load_dotenv
from pathlib import Path
import typing


def load_dotenv_if_available(file_path: typing.Optional[str] = None):
    env_file_path = file_path or Path(__file__).parent / ".env"
    load_dotenv(env_file_path)

aiser/env_tools/load_dotenv_if_available.py
penlight-ai/aiser
from .identifiable_entity import IdentifiableEntity

aiser/identifiable_entities/__init__.py
penlight-ai/aiser
import typing


class IdentifiableEntity:
    def __init__(self, entity_id: str):
        super().__init__()
        self._id = entity_id

    def accepts_id(self, entity_id: str) -> bool:
        return self._id == entity_id

    def get_id(self) -> typing.Optional[str]:
        return self._id


aiser/identifiable_entities/identifiable_entity.py
penlight-ai/aiser
from .knowledge_base import KnowledgeBase
from .semantic_search_result import SemanticSearchResult

aiser/knowledge_base/__init__.py
penlight-ai/aiser
import typing
from abc import ABC, abstractmethod
from typing import List

from ..identifiable_entities import IdentifiableEntity
from aiser.knowledge_base.semantic_search_result import SemanticSearchResult


class KnowledgeBase(IdentifiableEntity, ABC):
    def __init__(self, knowledge_base_id: str):
        super().__init__(entity_id=knowledge_base_id)

    @abstractmethod
    def perform_semantic_search(self, query_text: str, desired_number_of_results: int) -> List[SemanticSearchResult]:
        raise NotImplementedError

aiser/knowledge_base/knowledge_base.py
penlight-ai/aiser
from typing import Optional

from pydantic import BaseModel


class SemanticSearchResult(BaseModel):
    """
    Represents one result of a semantic search operation.

    Attributes:
        content (str): The content of the result.
        score (float): The confidence score obtained during similarity search.
    """

    content: str
    score: Optional[float] = None

aiser/knowledge_base/semantic_search_result.py
penlight-ai/aiser
from pydantic import BaseModel


class ChatMessage(BaseModel):
    text_content: str

aiser/models/chat_message.py
penlight-ai/aiser
from pydantic import BaseModel
from typing import List


class SemanticSearchRequest(BaseModel):
    text: str
    numResults: int


class SemanticSearchResultDto(BaseModel):
    content: str
    score: float


class SemanticSearchResultResponseDto(BaseModel):
    results: List[SemanticSearchResultDto]


class ChatMessageDto(BaseModel):
    textContent: str


class AgentChatRequest(BaseModel):
    messages: List[ChatMessageDto]


class AgentChatResponse(BaseModel):
    outputMessage: ChatMessageDto


class PublicKeyInfo(BaseModel):
    publicKey: str
    keyId: str


class VersionInfo(BaseModel):
    version: str

aiser/models/dtos.py
penlight-ai/aiser
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
import base64


def base64_to_pem(base64_key) -> str:
    decoded_key = base64.b64decode(base64_key)
    public_key = serialization.load_der_public_key(decoded_key, backend=default_backend())
    pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    return pem.decode()


def meets_minimum_version(current_version: str, min_version: str) -> bool:
    current_version_split = current_version.split(".")
    min_version_split = min_version.split(".")
    for current_version_part, min_version_part in zip(current_version_split, min_version_split):
        if int(current_version_part) < int(min_version_part):
            return False
    return True

aiser/utils/__init__.py
penlight-ai/aiser
import typing
import os
import asyncio
from langchain.chat_models import ChatOpenAI
from langchain.callbacks.base import BaseCallbackHandler
from langchain.schema import HumanMessage
from aiser import RestAiServer, Agent
from aiser.models import ChatMessage


class CustomCallbackHandler(BaseCallbackHandler):
    def __init__(self):
        self._tokens: typing.List[str] = []
        self._has_finished: bool = False

    def on_llm_new_token(self, token: str, **kwargs: typing.Any) -> None:
        self._tokens.append(token)

    def on_llm_end(self, response, **kwargs: typing.Any) -> None:
        self._has_finished = True

    async def get_generated_tokens(self) -> typing.AsyncGenerator[typing.List[str], None]:
        while (not self._has_finished) or (len(self._tokens) > 0):
            if len(self._tokens) > 0:
                yield self._tokens
                self._tokens = []
            await asyncio.sleep(0.05)


class PromptBasedAgent(Agent):
    def __init__(
            self,
            agent_id: str,
            starting_prompt: str,
            openai_api_key: str,
            model_name: str,
    ):
        super().__init__(agent_id=agent_id)
        self._starting_prompt = starting_prompt
        self._openai_api_key = openai_api_key
        self._model_name = model_name

    async def reply(self, messages: typing.List[ChatMessage]) -> typing.AsyncGenerator[ChatMessage, None]:
        callback_handler = CustomCallbackHandler()
        ai_model = self._make_model(
            callback_handler=callback_handler
        )
        model_input = self._make_model_input(messages=messages)
        model_execution_task = asyncio.create_task(self._execute_model(
            chat_model=ai_model,
            model_input=model_input
        ))

        async for tokens in callback_handler.get_generated_tokens():
            for token in tokens:
                yield ChatMessage(
                    text_content=token,
                )
        await model_execution_task

    def _make_model(self, callback_handler: CustomCallbackHandler) -> ChatOpenAI:
        return ChatOpenAI(
            openai_api_key=self._openai_api_key,
            model=self._model_name,
            streaming=True,
            temperature=0,
            callbacks=[callback_handler],
            verbose=True
        )

    def _make_model_input(self, messages: typing.List[ChatMessage]) -> str:
        starting_message = ChatMessage(
            text_content=self._starting_prompt
        )
        messages_combined = [starting_message] + list(messages)
        return self._messages_to_single_str(messages_combined)

    def _messages_to_single_str(self, messages: typing.List[ChatMessage]) -> str:
        final_prompt = ""
        for i, message in enumerate(messages):
            final_prompt += f"message {i}: {message.text_content}\n"
        final_prompt += f"message {len(messages)}:"
        return final_prompt

    async def _execute_model(self, chat_model, model_input: str):
        await chat_model.apredict_messages(messages=[
            HumanMessage(content=model_input)
        ])


if __name__ == '__main__':
    # Loading Environment Variables:
    # Duplicate the file sample.env and rename it to .env then fill in the values in the .env file.
    # Set the environment variable AISER_ENVIRONMENT to "Development" only for local development.
    # Otherwise, you can leave it empty or set it to "Production"
    from dotenv import load_dotenv

    load_dotenv()
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

    opposites_agent = PromptBasedAgent(
        agent_id='8d4ac909-ce3e-48c6-8233-63fa21da5b23',  # replace this with your agent id
        starting_prompt='''
                You are Opposite Agent. Whenever you receive a message from the user, 
                you should reply with the opposite of that message.
                Don't reply with anything other than the exact opposite of the user's message.
                ''',
        openai_api_key=OPENAI_API_KEY,
        model_name='gpt-3.5-turbo'
    )

    general_assistant_agent = PromptBasedAgent(
        agent_id='4b4b4bad-bdde-4cba-9170-ac8ae11d994a',  # replace this with your agent id
        starting_prompt='''
            You are a friendly and helpful AI assistant.
            ''',
        openai_api_key=OPENAI_API_KEY,
        model_name='gpt-3.5-turbo'
    )

    server = RestAiServer(
        agents=[
            opposites_agent,
            general_assistant_agent
        ],
        port=5000
    )
    server.run()

examples/langchain_agent/app.py
penlight-ai/aiser
from typing import List
import typing
from aiser import RestAiServer, KnowledgeBase, SemanticSearchResult, Agent
from aiser.models import ChatMessage
import asyncio


class KnowledgeBaseExample(KnowledgeBase):
    def perform_semantic_search(self, query_text: str, desired_number_of_results: int) -> List[SemanticSearchResult]:
        result_example = SemanticSearchResult(
            content="This is an example of a semantic search result",
            score=0.5,
        )
        return [result_example for _ in range(desired_number_of_results)]


class AgentExample(Agent):
    async def reply(self, messages: typing.List[ChatMessage]) -> typing.AsyncGenerator[ChatMessage, None]:
        reply_message = "This is an example of a reply from an agent"
        for character in reply_message:
            yield ChatMessage(text_content=character)
            await asyncio.sleep(0.1)


if __name__ == '__main__':
    server = RestAiServer(
        agents=[
            AgentExample(
                agent_id='10209b93-2dd0-47a0-8eb2-33fb018a783b'  # replace with your agent id
            ),
        ],
        knowledge_bases=[
            KnowledgeBaseExample(
                knowledge_base_id='85bc1c72-b8e0-4042-abcf-8eb2d478f207'  # replace with your knowledge base id
            ),
        ],
        port=5000
    )
    server.run()

examples/minimal_agent_and_knowledge_base/app.py
penlight-ai/aiser
from setuptools import setup, find_packages


def read_version():
    with open('aiser/version.py', 'r') as f:
        lines = f.readlines()
    for line in lines:
        if line.startswith('__version__'):
            delim = '"' if '"' in line else "'"
            return line.split(delim)[1]
    raise RuntimeError('Unable to find version string.')


version = read_version()

setup(
    name='aiser',
    version=version,
    url='https://github.com/penlight-ai/aiapi',
    author='Penlight AI Inc.',
    author_email='support@penlight.ai',
    description='Python package to serve AI agents and knowledge bases for Penlight AI.',
    long_description='Python package to serve AI agents and knowledge bases for Penlight AI.',
    packages=find_packages(),
    install_requires=[
        'uvicorn',
        'fastapi',
        'pyjwt[crypto]',
        'httpx',
    ],
    license='Apache License 2.0',
    classifiers=[
        'Development Status :: 1 - Planning',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: Apache Software License',
        'Programming Language :: Python :: 3.11',
    ],
)

setup.py
penlight-ai/aiser
import unittest
from aiser.utils import meets_minimum_version


class CurrentSemanticVersionMeetsMinVersionTestCase(unittest.TestCase):
    def assert_acceptable(self, current_version: str, min_version: str):
        self.assertTrue(meets_minimum_version(current_version, min_version))

    def assert_not_acceptable(self, current_version: str, min_version: str):
        self.assertFalse(meets_minimum_version(current_version, min_version))

    def test_low_major_version_is_not_accepted(self):
        self.assert_not_acceptable(current_version="0.1.0", min_version="1.0.0")

    def test_low_minor_version_is_not_accepted(self):
        self.assert_not_acceptable(current_version="1.0.0", min_version="1.1.0")

    def test_low_patch_version_is_not_accepted(self):
        self.assert_not_acceptable(current_version="1.1.0", min_version="1.1.1")

    def test_high_major_version_is_accepted(self):
        self.assert_acceptable(current_version="2.0.0", min_version="1.0.0")

    def test_high_minor_version_is_accepted(self):
        self.assert_acceptable(current_version="1.2.0", min_version="1.1.0")

    def test_high_patch_version_is_accepted(self):
        self.assert_acceptable(current_version="1.1.2", min_version="1.1.1")

    def test_equal_version_is_accepted(self):
        self.assert_acceptable(current_version="1.1.1", min_version="1.1.1")

tests/utils_tests.py
penlight-ai/aiser