get_systems - Get Systems Prefect Blocks

Enterprise-grade Prefect blocks for LLM operations, HTTP requests, and data models.

🐍 Available on PyPI — install it with pip install get-systems

Features

LLM Module (get_systems.llm)

• ✅ OpenAI & Azure OpenAI Support - Seamlessly switch between providers
• ✅ Environment Variables - Automatic fallback to env vars
• ✅ Retry Logic - Exponential backoff with jitter
• ✅ Caching - Optional in-memory response caching
• ✅ Safe Logging - No API key leaks
• ✅ Flexible Parameters - Pass any OpenAI API parameter via **kwargs
• ✅ Function Calling - Full support for tools

HTTP Module (get_systems.http)

• ✅ Multiple Auth Types - None, Basic, Token, Bearer
• ✅ Async HTTP Client - Built on httpx
• ✅ Prefect Integration - First-class block support

Models Module (get_systems.models)

• ✅ Pydantic Models - Type-safe data models with validation
• ✅ Address Parsing - German address format support with normalization
• ✅ Contact Management - Client, Debtor, and Contact models
• ✅ GDPR Compliance - Data protection and source tracking fields
• ✅ Bank Accounts - IBAN/BIC validation via python-stdnum
• ✅ Event Handling - Case events and interest calculations

Azure Content Understanding Module (get_systems.azure_cu)

• ✅ Async Azure SDK Client - URL analysis through Azure AI Content Understanding
• ✅ Prefect Block Config - Store endpoint, key, and optional API version
• ✅ Usage Metadata - Emits token usage metadata through Prefect events when available
• ✅ LangChain Tool Block - Optional separate block for langchain-azure-ai

Langfuse Module (get_systems.langfuse)

• ✅ Prefect Credentials Block - Store Langfuse keys and base URL in Prefect
• ✅ Client Factory - Create an authenticated client with get_client()

Installation

Install Everything (Recommended)

pip install "get_systems[all]"

Install Specific Modules

Install only what you need:

# For LLM operations only
pip install "get_systems[llm]"

# For HTTP operations only  
pip install "get_systems[http]"

# For Azure Content Understanding only
pip install "get_systems[azure_cu]"

# For Azure Content Understanding LangChain tools
pip install "get_systems[azure_cu_langchain]"

# For Langfuse credentials and client
pip install "get_systems[langfuse]"

# Install several modules
pip install "get_systems[llm,http,azure_cu]"

From Azure Artifacts

# All modules
pip install "get_systems[all]" --extra-index-url https://pkget_systems.dev.azure.com/get-systems/_packaging/get-systems/pypi/simple/

# Specific modules
pip install "get_systems[llm]" --extra-index-url https://pkget_systems.dev.azure.com/get-systems/_packaging/get-systems/pypi/simple/
pip install "get_systems[http]" --extra-index-url https://pkget_systems.dev.azure.com/get-systems/_packaging/get-systems/pypi/simple/

What Gets Installed

Quick Start

LLM Operations

from get_systems.llm import GptCompletionBlock, GptAuth, LlmRuntime
from prefect import flow

# Configure auth
auth = GptAuth(
    api_key="sk-...",
    model="gpt-4o-mini",
    is_azure=False
)

# Create completion block
block = GptCompletionBlock(
    auth=auth,
    prompt="What is Prefect?",
    temperature=0.7
)

@flow
async def my_llm_flow():
    result = await block.run()
    print(result.content)

HTTP Operations

from get_systems.http import HttpAuth, HttpBlock
from prefect import flow

# Configure HTTP auth
auth = HttpAuth(
    auth_type="bearer",
    token="your-token"
)

# Create HTTP block
http_block = HttpBlock(
    auth=auth,
    url="https://api.example.com"
)

@flow
async def my_http_flow():
    response = await http_block.request("GET", "/users")
    print(response.json())

Token obtained with username and password

For APIs such as Django REST Framework that issue a token from api-token-auth/, select Token authentication and enable Extract Token on the HTTP block. Configure auth_url, username, and password; the password is stored as a Prefect secret field. Before every request, the block posts {"username": "...", "password": "..."} to auth_url, reads the token field from the JSON response, and sends the target request with Authorization: Token <token>.

from get_systems.http import HttpAuth, HttpBlock

http_block = HttpBlock(
    url="https://api.example.com",
    extract_token=True,
    auth_url="/api-token-auth/",
    username="admin",
    password="use-a-prefect-secret-in-production",
    auth=HttpAuth(auth_type="token"),
)

response = await http_block.request("GET", "/users")
response.raise_for_status()

The retrieved token is used only for that request and is not saved in the Prefect block. auth_url may also be an absolute URL.

Azure Content Understanding

from get_systems.azure_cu import AzureAIContentBlock
from prefect import flow

cu_block = AzureAIContentBlock(
    endpoint="https://your-resource.services.ai.azure.com",
    key="your-key",
)

@flow
async def analyze_document():
    result = await cu_block.analyze_url(
        analyzer_id="auftrag",
        document_url="https://example.com/document.pdf",
    )
    print(result)

Azure Content Understanding LangChain Tool

from get_systems.azure_cu import AzureAIContentLangChainBlock

config_block = await AzureAIContentLangChainBlock.load("azure-cu-langchain-config")
tool = config_block.get_tool(analyzer_id="auftrag")

Data Models

from get_systems.models import Address, Debtor, ImportData, PersonType

# Parse German address from string
address = Address.from_raw("Zimmerstraße 456 c/o Amtsgericht, 63225 Langen, DE")
print(address.street)  # "Zimmerstraße 456 c/o Amtsgericht"
print(address.zip_code)  # "63225"
print(address.city)  # "Langen"

# Create address with validation
address = Address(
    street="Musterstraße 321a",
    zip_code="12345",
    city="Berlin",
    country_code="DE",
    validity="G"  # Gültig (Valid)
)

# GDPR-compliant data tracking
address_with_source = Address(
    street="Hauptstraße 10",
    zip_code="10115",
    city="Berlin",
    country_code="DE",
    source_contact_designation="Auskunftei",
    source_contact_id="12345",
    source_date="2026-06-01"
)

# Create debtor with contact info
debtor = Debtor(
    first_name="Max",
    name1="Mustermann",
    addresses=[address],
    person_type=PersonType.MALE,
)

# Build the top-level import payload
payload = ImportData(
    debtors=[debtor],
    robot_name="invoice-import",
    flow_run_id="01a05d2a-6d74-747b-a028-a168be25ef03",
    prefect_instance="prefect-prod1",
)

ImportData contains the top-level fields ccases, clients, debtors, bank_accounts, file_entries, robot_name, flow_run_id, prefect_instance, and inherited extra_fields.

Langfuse Credentials

from get_systems.langfuse import LangfuseCredentials

credentials = await LangfuseCredentials.load("langfuse-dev")
client = credentials.get_client()

Model Data Masking

Models derived from MaskedBaseModel or WithExtraFields can mark sensitive fields declaratively:

from pydantic import Field
from get_systems.models import MaskedBaseModel

class DebtorTrace(MaskedBaseModel):
    name: str | None = Field(
        default=None,
        json_schema_extra={"mask": True},
    )
    email: str | None = Field(
        default=None,
        json_schema_extra={"mask": True},
    )

original = DebtorTrace(
    name="Max Mustermann",
    email="max@example.com",
)
masked = original.masked_dump()  # Returns a copy with masked values
print(masked.name)  # "Max...n"
print(masked.email)  # "max...m"

masked_dump() returns a copy of the model instance with sensitive fields masked. All methods from the base class are preserved. Nested MaskedBaseModel instances are recursively masked. Sensitive strings are rendered as their first three characters, ..., and their last character, for example Max...n. Short values are rendered as ***. The original model remains unchanged.

Register Blocks in Prefect

# Register all blocks
prefect block register -m get_systems.llm.gpt_blocks
prefect block register -m get_systems.http
prefect block register -m get_systems.azure_cu
prefect block register -m get_systems.langfuse

# View registered blocks
prefect block ls

Import Styles

All import styles are supported:

# Submodule imports (recommended)
from get_systems.llm import GptCompletionBlock, GptAuth
from get_systems.http import HttpAuth, HttpBlock
from get_systems.azure_cu import AzureAIContentBlock, AzureAIContentLangChainBlock
from get_systems.langfuse import LangfuseCredentials

# Direct module imports
from get_systems.llm.gpt_blocks import GptCompletionBlock
from get_systems.http.http_block import HttpAuth
from get_systems.azure_cu.blocks import AzureAIContentBlock
from get_systems.langfuse.credentials import LangfuseCredentials

# Package-level imports
from get_systems import GptCompletionBlock, HttpAuth, AzureAIContentBlock, LangfuseCredentials

Environment Variables

OpenAI Configuration:

OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o-mini
OPENAI_BASE_URL=https://api.openai.com/v1  # optional

Azure OpenAI Configuration:

OPENAI_API_KEY=your-azure-key
OPENAI_BASE_URL=https://your-resource.openai.azure.com
OPENAI_MODEL=your-deployment-name
OPENAI_API_VERSION=2024-02-15-preview
OPENAI_IS_AZURE=true

Azure Content Understanding Configuration:

CONTENT_UNDERSTANDING_ENDPOINT=https://your-resource.services.ai.azure.com
CONTENT_UNDERSTANDING_KEY=your-key
CONTENT_UNDERSTANDING_API_VERSION=2025-11-01  # optional; SDK default is used when omitted

Documentation

• docs/QUICKSTART-GS.md - Quick start and usage examples
• docs/MODELS.md - Complete models documentation (Address, Contact, BankAccount, etc.)
• docs/MIGRATION.md - Migration guide from old packages
• LLM Module: Full OpenAI/Azure OpenAI integration with enterprise features
• HTTP Module: Flexible HTTP client with multiple authentication types
• Models Module: Pydantic models for addresses, contacts, bank accounts with validation
• Prefect Integration: Native Prefect block support for LLM and HTTP modules

License

Proprietary - Get Systems