Client API Reference¶

The client module provides a high-level API for fetching Gemini resources with support for TOFU certificate validation, redirects, and timeouts.

Overview¶

The Nauyaca client is built on Python's asyncio.Protocol and asyncio.Transport pattern for efficient, non-blocking I/O. It provides:

High-level async/await interface via GeminiClient
TOFU (Trust On First Use) validation for secure connections without CA infrastructure
Automatic redirect following with loop detection
Client certificate authentication for restricted resources
Configurable timeouts and connection settings

GeminiClient¶

GeminiClient ¶

GeminiClient(
    timeout: float = 30.0,
    max_redirects: int = MAX_REDIRECTS,
    ssl_context: SSLContext | None = None,
    verify_ssl: bool = False,
    trust_on_first_use: bool = True,
    tofu_db_path: Path | None = None,
    client_cert: Path | str | None = None,
    client_key: Path | str | None = None,
)

High-level Gemini client with async/await API.

This class provides a simple, high-level interface for getting Gemini resources. It handles connection management, TLS, redirects, and timeouts.

Examples:

>>> # Basic usage
>>> async with GeminiClient() as client:
...     response = await client.get('gemini://example.com/')
...     print(response.body)

>>> # With custom timeout and redirect settings
>>> client = GeminiClient(timeout=30, max_redirects=3)
>>> response = await client.get('gemini://example.com/')

>>> # Disable redirect following
>>> response = await client.get(
...     'gemini://example.com/',
...     follow_redirects=False
... )

Initialize the Gemini client.

Parameters:

Name	Type	Description	Default
`timeout`	`float`	Request timeout in seconds. Default is 30 seconds.	`30.0`
`max_redirects`	`int`	Maximum number of redirects to follow. Default is 5.	`MAX_REDIRECTS`
`ssl_context`	`SSLContext \| None`	Custom SSL context. If None, a default context will be created based on verify_ssl and trust_on_first_use settings.	`None`
`verify_ssl`	`bool`	Whether to verify SSL certificates using CA validation. Default is False. For Gemini, you should use TOFU instead.	`False`
`trust_on_first_use`	`bool`	Whether to use TOFU certificate validation. Default is True. This is the recommended mode for Gemini.	`True`
`tofu_db_path`	`Path \| None`	Path to TOFU database. If None, uses default location (~/.nauyaca/tofu.db).	`None`
`client_cert`	`Path \| str \| None`	Path to client certificate file (PEM format) for authentication with servers that require client certificates.	`None`
`client_key`	`Path \| str \| None`	Path to client private key file (PEM format). Required if client_cert is provided.	`None`

Source code in src/nauyaca/client/session.py

def __init__(
    self,
    timeout: float = 30.0,
    max_redirects: int = MAX_REDIRECTS,
    ssl_context: ssl.SSLContext | None = None,
    verify_ssl: bool = False,
    trust_on_first_use: bool = True,
    tofu_db_path: Path | None = None,
    client_cert: Path | str | None = None,
    client_key: Path | str | None = None,
):
    """Initialize the Gemini client.

    Args:
        timeout: Request timeout in seconds. Default is 30 seconds.
        max_redirects: Maximum number of redirects to follow. Default is 5.
        ssl_context: Custom SSL context. If None, a default context will be
            created based on verify_ssl and trust_on_first_use settings.
        verify_ssl: Whether to verify SSL certificates using CA validation.
            Default is False. For Gemini, you should use TOFU instead.
        trust_on_first_use: Whether to use TOFU certificate validation.
            Default is True. This is the recommended mode for Gemini.
        tofu_db_path: Path to TOFU database. If None, uses default location
            (~/.nauyaca/tofu.db).
        client_cert: Path to client certificate file (PEM format) for
            authentication with servers that require client certificates.
        client_key: Path to client private key file (PEM format). Required
            if client_cert is provided.
    """
    self.timeout = timeout
    self.max_redirects = max_redirects
    self.verify_ssl = verify_ssl
    self.trust_on_first_use = trust_on_first_use

    # Validate client cert/key pair
    if client_cert and not client_key:
        raise ValueError("client_key is required when client_cert is provided")
    if client_key and not client_cert:
        raise ValueError("client_cert is required when client_key is provided")

    # Initialize TOFU database if needed
    if self.trust_on_first_use:
        self.tofu_db: TOFUDatabase | None = TOFUDatabase(tofu_db_path)
    else:
        self.tofu_db = None

    # Create SSL context if not provided
    if ssl_context is None:
        if verify_ssl:
            # CA-based verification (not recommended for Gemini)
            self.ssl_context = create_client_context(
                verify_mode=ssl.CERT_REQUIRED,
                check_hostname=True,
                certfile=str(client_cert) if client_cert else None,
                keyfile=str(client_key) if client_key else None,
            )
        else:
            # TOFU mode or testing mode - accept all certificates
            # TOFU validation happens after connection is established
            self.ssl_context = create_client_context(
                verify_mode=ssl.CERT_NONE,
                check_hostname=False,
                certfile=str(client_cert) if client_cert else None,
                keyfile=str(client_key) if client_key else None,
            )
    else:
        self.ssl_context = ssl_context

aenter `async` ¶

__aenter__() -> GeminiClient

Async context manager entry.

Source code in src/nauyaca/client/session.py

async def __aenter__(self) -> "GeminiClient":
    """Async context manager entry."""
    return self

aexit `async` ¶

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None

Async context manager exit.

Source code in src/nauyaca/client/session.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> None:
    """Async context manager exit."""
    pass

delete `async` ¶

delete(
    url: str, token: str | None = None
) -> GeminiResponse

Delete a resource via zero-byte Titan upload.

In the Titan protocol, a zero-byte upload indicates deletion. The server may or may not support delete operations.

Parameters:

Name	Type	Description	Default
`url`	`str`	The target URL. Can be either gemini:// or titan:// scheme.	required
`token`	`str \| None`	Optional authentication token for the server.	`None`

Returns:

Type	Description
`GeminiResponse`	A GeminiResponse from the server indicating success or failure.

Raises:

Type	Description
`ValueError`	If the URL is invalid.
`TimeoutError`	If the request times out.
`ConnectionError`	If the connection fails.

Examples:

>>> # Delete a resource
>>> response = await client.delete(
...     'gemini://example.com/uploads/old-file.gmi',
...     token='secret-token',
... )
>>> if response.is_success():
...     print("Resource deleted")

Source code in src/nauyaca/client/session.py

async def delete(
    self,
    url: str,
    token: str | None = None,
) -> GeminiResponse:
    """Delete a resource via zero-byte Titan upload.

    In the Titan protocol, a zero-byte upload indicates deletion.
    The server may or may not support delete operations.

    Args:
        url: The target URL. Can be either gemini:// or titan:// scheme.
        token: Optional authentication token for the server.

    Returns:
        A GeminiResponse from the server indicating success or failure.

    Raises:
        ValueError: If the URL is invalid.
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> # Delete a resource
        >>> response = await client.delete(
        ...     'gemini://example.com/uploads/old-file.gmi',
        ...     token='secret-token',
        ... )
        >>> if response.is_success():
        ...     print("Resource deleted")
    """
    return await self.upload(url, b"", mime_type="text/gemini", token=token)

get `async` ¶

get(
    url: str, follow_redirects: bool = True
) -> GeminiResponse

Get a Gemini resource.

Parameters:

Name	Type	Description	Default
`url`	`str`	The Gemini URL to get.	required
`follow_redirects`	`bool`	Whether to automatically follow redirects. Default is True.	`True`

Returns:

Type	Description
`GeminiResponse`	A GeminiResponse object with status, meta, and optional body.

Raises:

Type	Description
`ValueError`	If the URL is invalid.
`TimeoutError`	If the request times out.
`ConnectionError`	If the connection fails.

Examples:

>>> response = await client.get('gemini://example.com/')
>>> if response.is_success():
...     print(response.body)

Source code in src/nauyaca/client/session.py

async def get(
    self,
    url: str,
    follow_redirects: bool = True,
) -> GeminiResponse:
    """Get a Gemini resource.

    Args:
        url: The Gemini URL to get.
        follow_redirects: Whether to automatically follow redirects.
            Default is True.

    Returns:
        A GeminiResponse object with status, meta, and optional body.

    Raises:
        ValueError: If the URL is invalid.
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> response = await client.get('gemini://example.com/')
        >>> if response.is_success():
        ...     print(response.body)
    """
    # Validate URL
    validate_url(url)

    # Get with redirect following if enabled
    if follow_redirects:
        return await self._get_with_redirects(url, max_redirects=self.max_redirects)
    else:
        return await self._get_single(url)

upload `async` ¶

upload(
    url: str,
    content: bytes | str,
    mime_type: str = "text/gemini",
    token: str | None = None,
) -> GeminiResponse

Upload content to a Gemini server via the Titan protocol.

Titan is Gemini's upload companion protocol. It uses the same port (1965) and TLS requirements as Gemini, but allows uploading content.

Parameters:

Name	Type	Description	Default
`url`	`str`	The target URL. Can be either gemini:// or titan:// scheme. If gemini://, it will be converted to titan://.	required
`content`	`bytes \| str`	The content to upload. Can be bytes or str (will be encoded as UTF-8).	required
`mime_type`	`str`	MIME type of the content. Default is "text/gemini".	`'text/gemini'`
`token`	`str \| None`	Optional authentication token for the server.	`None`

Returns:

Type	Description
`GeminiResponse`	A GeminiResponse from the server indicating success or failure.

Raises:

Type	Description
`ValueError`	If the URL is invalid.
`TimeoutError`	If the request times out.
`ConnectionError`	If the connection fails.

Examples:

>>> # Upload text content
>>> response = await client.upload(
...     'gemini://example.com/uploads/note.gmi',
...     '# My Note\n\nHello, Geminispace!',
... )

>>> # Upload binary content
>>> with open('image.png', 'rb') as f:
...     response = await client.upload(
...         'gemini://example.com/uploads/image.png',
...         f.read(),
...         mime_type='image/png',
...     )

>>> # Upload with authentication
>>> response = await client.upload(
...     'gemini://example.com/uploads/file.txt',
...     'Hello!',
...     token='secret-token',
... )

Source code in src/nauyaca/client/session.py

async def upload(
    self,
    url: str,
    content: bytes | str,
    mime_type: str = "text/gemini",
    token: str | None = None,
) -> GeminiResponse:
    """Upload content to a Gemini server via the Titan protocol.

    Titan is Gemini's upload companion protocol. It uses the same port (1965)
    and TLS requirements as Gemini, but allows uploading content.

    Args:
        url: The target URL. Can be either gemini:// or titan:// scheme.
            If gemini://, it will be converted to titan://.
        content: The content to upload. Can be bytes or str (will be
            encoded as UTF-8).
        mime_type: MIME type of the content. Default is "text/gemini".
        token: Optional authentication token for the server.

    Returns:
        A GeminiResponse from the server indicating success or failure.

    Raises:
        ValueError: If the URL is invalid.
        asyncio.TimeoutError: If the request times out.
        ConnectionError: If the connection fails.

    Examples:
        >>> # Upload text content
        >>> response = await client.upload(
        ...     'gemini://example.com/uploads/note.gmi',
        ...     '# My Note\\n\\nHello, Geminispace!',
        ... )

        >>> # Upload binary content
        >>> with open('image.png', 'rb') as f:
        ...     response = await client.upload(
        ...         'gemini://example.com/uploads/image.png',
        ...         f.read(),
        ...         mime_type='image/png',
        ...     )

        >>> # Upload with authentication
        >>> response = await client.upload(
        ...     'gemini://example.com/uploads/file.txt',
        ...     'Hello!',
        ...     token='secret-token',
        ... )
    """
    # Convert content to bytes if string
    if isinstance(content, str):
        content_bytes = content.encode("utf-8")
    else:
        content_bytes = content

    # Convert gemini:// to titan:// if needed
    if url.startswith("gemini://"):
        titan_url_base = "titan://" + url[9:]
    elif url.startswith("titan://"):
        titan_url_base = url
    else:
        raise ValueError("URL must use gemini:// or titan:// scheme")

    # Build Titan URL with parameters
    # Format: titan://host/path;size=N;mime=TYPE;token=TOKEN
    titan_url = f"{titan_url_base};size={len(content_bytes)};mime={mime_type}"
    if token:
        titan_url += f";token={token}"

    # Parse URL to get host and port (use the base URL without params)
    parsed = parse_url(titan_url_base.replace("titan://", "gemini://"))

    # Get event loop
    loop = asyncio.get_running_loop()

    # Create future for response
    response_future: asyncio.Future = loop.create_future()

    # Create protocol instance
    protocol = TitanClientProtocol(titan_url, content_bytes, response_future)

    # Create connection using Protocol/Transport pattern
    try:
        transport, protocol = await asyncio.wait_for(
            loop.create_connection(
                lambda: protocol,
                host=parsed.hostname,
                port=parsed.port,
                ssl=self.ssl_context,
                server_hostname=parsed.hostname,
            ),
            timeout=self.timeout,
        )
    except TimeoutError as e:
        raise TimeoutError(f"Connection timeout: {url}") from e
    except OSError as e:
        raise ConnectionError(f"Connection failed: {e}") from e

    try:
        # If TOFU is enabled, verify the certificate
        if self.tofu_db:
            cert = protocol.get_peer_certificate()
            if cert:
                is_valid, message = self.tofu_db.verify(
                    parsed.hostname, parsed.port, cert
                )

                if not is_valid and message == "changed":
                    # Certificate changed - get old info and raise error
                    old_info = self.tofu_db.get_host_info(
                        parsed.hostname, parsed.port
                    )
                    old_fingerprint = (
                        old_info["fingerprint"] if old_info else "unknown"
                    )
                    new_fingerprint = get_certificate_fingerprint(cert)
                    raise CertificateChangedError(
                        parsed.hostname,
                        parsed.port,
                        old_fingerprint,
                        new_fingerprint,
                    )
                elif message == "first_use":
                    # First time seeing this host - trust it
                    self.tofu_db.trust(parsed.hostname, parsed.port, cert)

        # Wait for response with timeout
        response: GeminiResponse = await asyncio.wait_for(
            response_future, timeout=self.timeout
        )
        return response
    except TimeoutError as e:
        raise TimeoutError(f"Upload timeout: {url}") from e
    finally:
        # Ensure transport is closed
        transport.close()

Basic Usage¶

The simplest way to fetch a Gemini resource:

import asyncio
from nauyaca.client.session import GeminiClient

async def main():
    async with GeminiClient() as client:
        response = await client.get('gemini://geminiprotocol.net/')

        if response.is_success():
            print(response.body)
        else:
            print(f"Error {response.status}: {response.meta}")

asyncio.run(main())

Configuration Options¶

Create a client with custom settings:

from pathlib import Path
from nauyaca.client.session import GeminiClient

async def main():
    # Client with custom timeout and redirect settings
    client = GeminiClient(
        timeout=60.0,              # 60 second timeout
        max_redirects=3,            # Follow up to 3 redirects
        trust_on_first_use=True,    # Enable TOFU (recommended)
        tofu_db_path=Path("~/.config/nauyaca/tofu.db").expanduser()
    )

    response = await client.get('gemini://example.com/path')

Client Certificate Authentication¶

For servers that require client certificates (status 6x):

from pathlib import Path
from nauyaca.client.session import GeminiClient

async def main():
    client = GeminiClient(
        client_cert=Path("/path/to/client-cert.pem"),
        client_key=Path("/path/to/client-key.pem")
    )

    # Now requests will include your client certificate
    response = await client.get('gemini://restricted.example.com/')

Error Handling¶

Handle common error conditions:

import asyncio
from nauyaca.client.session import GeminiClient
from nauyaca.security.tofu import CertificateChangedError

async def main():
    async with GeminiClient() as client:
        try:
            response = await client.get('gemini://example.com/')

            if response.is_success():
                print(f"Content-Type: {response.mime_type}")
                print(response.body)
            elif response.is_redirect():
                print(f"Redirects to: {response.redirect_url}")
            else:
                print(f"Error {response.status}: {response.meta}")

        except CertificateChangedError as e:
            # Certificate changed - potential security issue
            print(f"WARNING: Certificate changed for {e.hostname}:{e.port}")
            print(f"Old fingerprint: {e.old_fingerprint}")
            print(f"New fingerprint: {e.new_fingerprint}")
            # User should verify this is legitimate before trusting

        except asyncio.TimeoutError:
            print("Request timed out")

        except ConnectionError as e:
            print(f"Connection failed: {e}")

        except ValueError as e:
            print(f"Invalid URL or redirect: {e}")

Uploading Content (Titan)¶

Upload content to a Gemini server using the Titan protocol:

async def main():
    async with GeminiClient() as client:
        # Upload text content
        response = await client.upload(
            'gemini://example.com/wiki/page.gmi',
            '# My Page\n\nContent here...',
            mime_type='text/gemini',
            token='auth-token',
        )

        if response.is_success():
            print("Upload successful!")

        # Upload binary content
        with open('image.png', 'rb') as f:
            response = await client.upload(
                'gemini://example.com/images/photo.png',
                f.read(),
                mime_type='image/png',
                token='auth-token',
            )

Deleting Content (Titan)¶

Delete a resource using a zero-byte Titan upload:

async def main():
    async with GeminiClient() as client:
        response = await client.delete(
            'gemini://example.com/wiki/old-page.gmi',
            token='auth-token',
        )

        if response.is_success():
            print("Deleted!")

Server Support Required

The server must have Titan enabled with enable_delete = true for delete operations to succeed.

Disabling Redirects¶

Sometimes you want to handle redirects manually:

async def main():
    async with GeminiClient() as client:
        # Don't follow redirects automatically
        response = await client.get(
            'gemini://example.com/',
            follow_redirects=False
        )

        if response.is_redirect():
            print(f"Got redirect to: {response.redirect_url}")
            # Decide whether to follow it yourself

GeminiResponse¶

GeminiResponse `dataclass` ¶

GeminiResponse(
    status: int,
    meta: str,
    body: str | bytes | None = None,
    url: str | None = None,
)

Represents a Gemini protocol response.

Attributes:

Name	Type	Description
`status`	`int`	Two-digit status code (10-69).
`meta`	`str`	Status-dependent metadata string. For success (2x), this is the MIME type. For redirects (3x), this is the redirect URL. For errors, this is an error message. For input (1x), this is the prompt.
`body`	`str \| bytes \| None`	Response body content (only present for 2x success responses). For text/* MIME types, this is a decoded string. For binary MIME types (images, audio, etc.), this is raw bytes.
`url`	`str \| None`	The URL this response came from (useful for tracking redirects).

Examples:

>>> response = GeminiResponse(
...     status=20,
...     meta='text/gemini',
...     body='# Hello World\nWelcome to Gemini!',
...     url='gemini://example.com/'
... )
>>> response.is_success()
True
>>> response.mime_type
'text/gemini'

charset `property` ¶

charset: str

Extract charset from MIME type parameters, defaulting to utf-8.

Returns:

Type	Description
`str`	The charset specified in the meta field, or 'utf-8' if not specified.

mime_type `property` ¶

mime_type: str | None

Get the MIME type from a success response.

Returns:

Type	Description
`str \| None`	The MIME type if this is a success response, None otherwise.

redirect_url `property` ¶

redirect_url: str | None

Get the redirect URL from a redirect response.

Returns:

Type	Description
`str \| None`	The redirect URL if this is a redirect response, None otherwise.

str ¶

__str__() -> str

Return a human-readable string representation of the response.

Source code in src/nauyaca/protocol/response.py

def __str__(self) -> str:
    """Return a human-readable string representation of the response."""
    lines = [f"Status: {self.status} - {self.meta}"]
    if self.url:
        lines.append(f"URL: {self.url}")
    if self.body:
        lines.append(f"Body: {len(self.body)} bytes")
    return "\n".join(lines)

is_redirect ¶

is_redirect() -> bool

Check if this response indicates a redirect (3x status code).

Source code in src/nauyaca/protocol/response.py

def is_redirect(self) -> bool:
    """Check if this response indicates a redirect (3x status code)."""
    return is_redirect(self.status)

is_success ¶

is_success() -> bool

Check if this response indicates success (2x status code).

Source code in src/nauyaca/protocol/response.py

def is_success(self) -> bool:
    """Check if this response indicates success (2x status code)."""
    return is_success(self.status)

Checking Response Types¶

The GeminiResponse class provides convenient methods for checking response status:

response = await client.get('gemini://example.com/')

# Check if request succeeded
if response.is_success():
    # Status 20-29: response has body content
    print(f"MIME type: {response.mime_type}")
    print(f"Body: {response.body}")

# Check if it's a redirect
elif response.is_redirect():
    # Status 30-39: meta contains redirect URL
    print(f"Redirect to: {response.redirect_url}")

# Otherwise it's an error or input request
else:
    status_category = interpret_status(response.status)
    print(f"{status_category}: {response.meta}")

Accessing Response Attributes¶

All response data is available as attributes:

response = await client.get('gemini://example.com/')

# Core attributes
print(f"Status: {response.status}")       # e.g., 20
print(f"Meta: {response.meta}")           # e.g., "text/gemini"
print(f"Body: {response.body}")           # Only present for 2x status
print(f"URL: {response.url}")             # The URL that was requested

# Convenience properties
print(f"MIME type: {response.mime_type}") # Extracted from meta (success only)
print(f"Charset: {response.charset}")     # Defaults to utf-8
print(f"Redirect: {response.redirect_url}") # Extracted from meta (redirect only)

Status Code Utilities¶

Utility functions for interpreting status codes:

interpret_status ¶

interpret_status(status: int) -> str

Interpret a status code and return its category name.

Parameters:

Name	Type	Description	Default
`status`	`int`	A two-digit Gemini status code (10-69).	required

Returns:

Type	Description
`str`	A string describing the general category of the status code.

Examples:

>>> interpret_status(20)
'SUCCESS'
>>> interpret_status(51)
'PERMANENT FAILURE'
>>> interpret_status(30)
'REDIRECT'

Source code in src/nauyaca/protocol/status.py

def interpret_status(status: int) -> str:
    """Interpret a status code and return its category name.

    Args:
        status: A two-digit Gemini status code (10-69).

    Returns:
        A string describing the general category of the status code.

    Examples:
        >>> interpret_status(20)
        'SUCCESS'
        >>> interpret_status(51)
        'PERMANENT FAILURE'
        >>> interpret_status(30)
        'REDIRECT'
    """
    if 10 <= status < 20:
        return "INPUT"
    elif 20 <= status < 30:
        return "SUCCESS"
    elif 30 <= status < 40:
        return "REDIRECT"
    elif 40 <= status < 50:
        return "TEMPORARY FAILURE"
    elif 50 <= status < 60:
        return "PERMANENT FAILURE"
    elif 60 <= status < 70:
        return "CLIENT CERTIFICATE REQUIRED"
    else:
        return "UNKNOWN"

is_success ¶

is_success(status: int) -> bool

Check if a status code indicates success (2x).

Parameters:

Name	Type	Description	Default
`status`	`int`	A two-digit Gemini status code.	required

Returns:

Type	Description
`bool`	True if the status code is in the success range (20-29).

Source code in src/nauyaca/protocol/status.py

def is_success(status: int) -> bool:
    """Check if a status code indicates success (2x).

    Args:
        status: A two-digit Gemini status code.

    Returns:
        True if the status code is in the success range (20-29).
    """
    return 20 <= status < 30

is_redirect ¶

is_redirect(status: int) -> bool

Check if a status code indicates a redirect (3x).

Parameters:

Name	Type	Description	Default
`status`	`int`	A two-digit Gemini status code.	required

Returns:

Type	Description
`bool`	True if the status code is in the redirect range (30-39).

Source code in src/nauyaca/protocol/status.py

def is_redirect(status: int) -> bool:
    """Check if a status code indicates a redirect (3x).

    Args:
        status: A two-digit Gemini status code.

    Returns:
        True if the status code is in the redirect range (30-39).
    """
    return 30 <= status < 40

is_input_required ¶

is_input_required(status: int) -> bool

Check if a status code indicates input is required (1x).

Parameters:

Name	Type	Description	Default
`status`	`int`	A two-digit Gemini status code.	required

Returns:

Type	Description
`bool`	True if the status code is in the input range (10-19).

Source code in src/nauyaca/protocol/status.py

def is_input_required(status: int) -> bool:
    """Check if a status code indicates input is required (1x).

    Args:
        status: A two-digit Gemini status code.

    Returns:
        True if the status code is in the input range (10-19).
    """
    return 10 <= status < 20

is_error ¶

is_error(status: int) -> bool

Check if a status code indicates an error (4x, 5x, or 6x).

Parameters:

Name	Type	Description	Default
`status`	`int`	A two-digit Gemini status code.	required

Returns:

Type	Description
`bool`	True if the status code indicates any type of error.

Source code in src/nauyaca/protocol/status.py

def is_error(status: int) -> bool:
    """Check if a status code indicates an error (4x, 5x, or 6x).

    Args:
        status: A two-digit Gemini status code.

    Returns:
        True if the status code indicates any type of error.
    """
    return 40 <= status < 70

Status Code Examples¶

from nauyaca.protocol.status import (
    interpret_status,
    is_success,
    is_redirect,
    is_input_required,
    is_error
)

response = await client.get('gemini://example.com/')

# Get category name
category = interpret_status(response.status)  # "SUCCESS", "REDIRECT", etc.

# Check specific categories
if is_success(response.status):
    print("Success!")
elif is_redirect(response.status):
    print(f"Redirect to: {response.redirect_url}")
elif is_input_required(response.status):
    print(f"Server needs input: {response.meta}")
elif is_error(response.status):
    print(f"Error: {response.meta}")

Common Patterns¶

Following Redirects Manually¶

async def fetch_with_manual_redirects(client, url, max_redirects=5):
    """Fetch a URL and manually handle redirects."""
    redirects_followed = 0
    current_url = url

    while redirects_followed < max_redirects:
        response = await client.get(current_url, follow_redirects=False)

        if not response.is_redirect():
            return response

        # Check if it's a gemini:// redirect
        redirect_url = response.redirect_url
        if not redirect_url.startswith('gemini://'):
            print(f"Warning: non-Gemini redirect to {redirect_url}")
            return response

        print(f"Following redirect to: {redirect_url}")
        current_url = redirect_url
        redirects_followed += 1

    raise ValueError(f"Too many redirects (>{max_redirects})")

Handling Input Prompts¶

from nauyaca.protocol.status import is_input_required, StatusCode

async def interactive_fetch(client, url):
    """Fetch a URL and handle input prompts interactively."""
    response = await client.get(url)

    # Check if server is requesting input
    if is_input_required(response.status):
        # Display the prompt to the user
        print(f"Server prompt: {response.meta}")

        # Get user input
        if response.status == StatusCode.SENSITIVE_INPUT:
            # Don't echo for sensitive input (like passwords)
            import getpass
            user_input = getpass.getpass("Input (hidden): ")
        else:
            user_input = input("Input: ")

        # Build URL with query string
        # Gemini uses '?' to separate path from query
        query_url = f"{url}?{user_input}"

        # Make request with user's input
        response = await client.get(query_url)

    return response

Client Certificate Authentication¶

async def fetch_with_cert(url, cert_path, key_path):
    """Fetch a resource using client certificate authentication."""
    from pathlib import Path

    client = GeminiClient(
        client_cert=Path(cert_path),
        client_key=Path(key_path)
    )

    try:
        response = await client.get(url)

        # Check for certificate-related errors
        if response.status == 60:
            print("Server requires a client certificate")
        elif response.status == 61:
            print("Certificate not authorized for this resource")
        elif response.status == 62:
            print("Certificate not valid")
        else:
            return response

    finally:
        # Client will be cleaned up automatically
        pass

TOFU Certificate Management¶

from nauyaca.security.tofu import TOFUDatabase, CertificateChangedError
from pathlib import Path

async def safe_fetch_with_tofu(url):
    """Fetch with TOFU validation and user confirmation on changes."""
    client = GeminiClient(trust_on_first_use=True)

    try:
        response = await client.get(url)
        return response

    except CertificateChangedError as e:
        # Certificate changed - ask user to verify
        print(f"\nWARNING: Certificate changed for {e.hostname}:{e.port}")
        print(f"Old fingerprint: {e.old_fingerprint}")
        print(f"New fingerprint: {e.new_fingerprint}")
        print("\nThis could be a legitimate certificate renewal,")
        print("or it could indicate a man-in-the-middle attack.")

        answer = input("\nDo you want to trust the new certificate? (yes/no): ")

        if answer.lower() == 'yes':
            # Trust the new certificate
            tofu_db = TOFUDatabase()
            # Need to fetch again to get the certificate
            # This time we'll trust it
            # (In practice, you'd want to extract and trust the cert directly)
            print("Please verify the fingerprint through a separate channel!")
        else:
            print("Certificate not trusted. Aborting.")
            raise

Batch Requests¶

async def fetch_multiple(urls):
    """Fetch multiple URLs concurrently."""
    async with GeminiClient() as client:
        # Create tasks for all URLs
        tasks = [client.get(url) for url in urls]

        # Run them concurrently
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        # Process results
        for url, response in zip(urls, responses):
            if isinstance(response, Exception):
                print(f"{url}: Error - {response}")
            elif response.is_success():
                print(f"{url}: Success - {len(response.body)} bytes")
            else:
                print(f"{url}: Status {response.status}")

Client API Reference¶

Overview¶

GeminiClient¶

GeminiClient ¶

__aenter__ async ¶

__aexit__ async ¶

delete async ¶

get async ¶

upload async ¶

Basic Usage¶

Configuration Options¶

Client Certificate Authentication¶

Error Handling¶

Uploading Content (Titan)¶

Deleting Content (Titan)¶

Disabling Redirects¶

GeminiResponse¶

GeminiResponse dataclass ¶

charset property ¶

mime_type property ¶

redirect_url property ¶

__str__ ¶

is_redirect ¶

is_success ¶

Checking Response Types¶

Accessing Response Attributes¶

Status Code Utilities¶

interpret_status ¶

is_success ¶

is_redirect ¶

is_input_required ¶

is_error ¶

Status Code Examples¶

Common Patterns¶

Following Redirects Manually¶

Handling Input Prompts¶

Client Certificate Authentication¶

TOFU Certificate Management¶

Batch Requests¶

See Also¶

aenter `async` ¶

aexit `async` ¶

delete `async` ¶

get `async` ¶

upload `async` ¶

GeminiResponse `dataclass` ¶

charset `property` ¶

mime_type `property` ¶

redirect_url `property` ¶

str ¶