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())

Signing request bodies in HttpBlock

Set the optional signing_secret (whsec_<base64>) on HttpBlock and every request() is signed per Standard Webhooks: webhook-id, webhook-timestamp and webhook-signature headers are added and the exact signed bytes are sent. Only json= and content= bodies are supported (data=/files= raise ValueError).

block = HttpBlock(url="https://hermes.example.com", signing_secret="whsec_...")
await block.request("POST", "/webhooks/x", json={"a": 1})

Signed Webhook (Standard Webhooks) for Prefect automations

SignedWebhook subclasses Prefect's built-in Webhook block, so it appears in the automation action "Call a webhook". It sends the request signed per the Standard Webhooks spec (HMAC-SHA256).

Block fields: url, method (default POST), headers, allow_private_urls, verify (as in Webhook) plus secret - the signing secret whsec_<base64>. The block adds webhook-id, webhook-timestamp and webhook-signature (v1,<base64>) headers and content-type: application/json (unless set). The exact signed bytes are sent: a string payload is sent as-is, a dict as compact JSON, None as an empty body.

from get_systems.http import SignedWebhook

block = SignedWebhook(url="https://hermes.example.com/webhooks/x", secret="whsec_...")
await block.save("hermes-signed")

Register the block type (see DEPLOYMENT.md):

prefect block register -m get_systems.http.signed_webhook

Note: the "Call a webhook" action is executed by the Prefect server, so the get-systems package (with this class) must be installed in the server image; otherwise the action fails with "The webhook block was invalid".

Token obtained with username and password

For APIs such as Django REST Framework that issue a token from api-token-auth/, select Token Login authentication. Configure the authentication URL, username, and password in that tab; the password is stored as a Prefect secret field. Before every request, the block posts {"username": "...", "password": "..."} to the configured 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",
    auth=HttpAuth(
        auth_type="token_login",
        auth_url="/api-token-auth/",
        username="admin",
        password="use-a-prefect-secret-in-production",
    ),
)

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