MagickMind
PythonResources

Corpus (v1)

Reference for magick_mind.resources.v1.corpus.

magick_mind.resources.v1.corpus

Corpus resource for Magick Mind SDK v1 API.

Provides methods for managing corpus (knowledge base) resources.

CorpusResourceV1

class CorpusResourceV1(BaseResource)

Resource client for corpus operations.

CorpusResourceV1.create

async def create(name: str,
                 description: str,
                 artifact_ids: Optional[list[str]] = None) -> Corpus

Create a new corpus.

Arguments:

  • name - Corpus name
  • description - Corpus description
  • artifact_ids - Optional list of artifact IDs to include

Returns:

Corpus object

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.get

async def get(corpus_id: str) -> Corpus

Get a corpus by ID.

Arguments:

  • corpus_id - The corpus ID

Returns:

Corpus object

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.list

async def list(*,
               cursor: Optional[str] = None,
               limit: Optional[int] = None,
               order: Optional[str] = None,
               search: Optional[str] = None,
               end_user_id: Optional[str] = None) -> ListCorpusResponse

List corpus, with optional filtering and pagination.

Arguments:

  • cursor - Pagination cursor (opaque string from PageInfo.cursors.after/before)
  • limit - Maximum number of results per page (default 20, max 100)
  • order - Sort order — "asc" or "desc" (default: asc)
  • search - Free-text search filter
  • end_user_id - Filter by end-user ID

Returns:

ListCorpusResponse with list of corpus and pagination info

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.update

async def update(corpus_id: str, name: str, description: str,
                 artifact_ids: list[str]) -> Corpus

Update an existing corpus.

Arguments:

  • corpus_id - The corpus ID to update
  • name - New corpus name
  • description - New corpus description
  • artifact_ids - New list of artifact IDs

Returns:

Corpus object

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.delete

async def delete(corpus_id: str) -> None

Delete a corpus.

Arguments:

  • corpus_id - The corpus ID to delete

Returns:

None (The API returns 204 No Content)

Example:

await client.v1.corpus.delete(corpus_id="corpus-123") print("Corpus deleted successfully")

Raises:

  • ProblemDetailsException - If request fails

CorpusResourceV1.add_artifact

async def add_artifact(corpus_id: str,
                       artifact_id: str,
                       *,
                       api_key: Optional[str] = None) -> AddArtifactsResponse

Add a single artifact to corpus and trigger ingestion.

Arguments:

  • corpus_id - The corpus ID
  • artifact_id - The artifact ID to add
  • api_key - Optional API key (sent as x-api-key header). Required for corpus-scoped operations on dev/prod.

Returns:

AddArtifactsResponse with result

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.add_artifacts

async def add_artifacts(corpus_id: str,
                        artifact_ids: list[str],
                        *,
                        api_key: Optional[str] = None) -> AddArtifactsResponse

Add multiple artifacts to corpus (max 20) and trigger batch ingestion.

Note: Only text-based formats (text/*, JSON, XML) and PDF are supported.

Arguments:

  • corpus_id - The corpus ID
  • artifact_ids - List of artifact IDs to add (max 20)
  • api_key - Optional API key (sent as x-api-key header). Required for corpus-scoped operations on dev/prod.

Returns:

AddArtifactsResponse with result

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.remove_artifact

async def remove_artifact(corpus_id: str, artifact_id: str) -> None

Remove artifact from corpus.

Arguments:

  • corpus_id - The corpus ID
  • artifact_id - The artifact ID to remove

Returns:

None

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.list_artifacts

async def list_artifacts(corpus_id: str,
                         cursor: Optional[str] = None,
                         limit: int = 20) -> list[CorpusArtifactItem]

List artifacts in a corpus with their ingestion statuses.

Returns a combined view of each artifact and its current ingestion state, suitable for polling or displaying corpus contents.

Arguments:

  • corpus_id - The corpus ID
  • cursor - Pagination cursor (opaque string from PageInfo.cursors.after/before)
  • limit - Maximum number of results per page (default 20, max 100)

Returns:

List of CorpusArtifactItem objects (artifact + ingestion status)

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.list_ingestion

async def list_ingestion(
        corpus_id: str,
        artifact_ids: Optional[list[str]] = None) -> list[IngestionStatus]

List ingestion statuses for artifacts in a corpus.

Arguments:

  • corpus_id - The corpus ID
  • artifact_ids - Optional list of specific artifact IDs to filter

Returns:

List of IngestionStatus objects

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.get_ingestion

async def get_ingestion(corpus_id: str, artifact_id: str) -> IngestionStatus

Get ingestion status for a single artifact.

Arguments:

  • corpus_id - The corpus ID
  • artifact_id - The artifact ID

Returns:

IngestionStatus object

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.get_artifact_status

async def get_artifact_status(corpus_id: str,
                              artifact_id: str) -> ArtifactStatus

Get ingestion status for a single artifact.

.. deprecated:: Use :meth:get_ingestion instead, which returns the updated IngestionStatus model.

Arguments:

  • corpus_id - The corpus ID
  • artifact_id - The artifact ID

Returns:

ArtifactStatus object

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.list_artifact_statuses

async def list_artifact_statuses(
        corpus_id: str,
        artifact_ids: Optional[list[str]] = None) -> list[ArtifactStatus]

List ingestion statuses for artifacts in corpus.

.. deprecated:: Use :meth:list_ingestion instead, which returns the updated IngestionStatus model.

Arguments:

  • corpus_id - The corpus ID
  • artifact_ids - Optional list of specific artifact IDs to filter

Returns:

List of ArtifactStatus objects

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.query

async def query(corpus_id: str,
                *,
                query: str,
                mode: Optional[str] = None,
                only_need_context: bool = False,
                enable_rerank: Optional[bool] = None,
                api_key: Optional[str] = None) -> QueryCorpusResponse

Query a corpus using semantic search.

Arguments:

  • corpus_id - The corpus ID to query
  • query - The search query text
  • mode - Query mode — one of naive|local|global|hybrid. Omit to let the server choose (defaults to hybrid).
  • only_need_context - If True, return raw context without LLM synthesis.
  • enable_rerank - Override rerank behavior. None = server default (on if provider configured), True = force on, False = force off.
  • api_key - Optional API key (sent as x-api-key header). Required for corpus-scoped operations on dev/prod.

Returns:

QueryCorpusResponse with the result text

Raises:

  • ProblemDetailsException - If the request fails

CorpusResourceV1.ingest

async def ingest(corpus_id: str,
                 *,
                 artifact_id: str,
                 api_key: Optional[str] = None) -> IngestResult

Add an already-uploaded artifact to a corpus and return an IngestResult.

This is a thin wrapper around :meth:add_artifact that returns the consolidated :class:~magick_mind.models.v1.corpus.IngestResult type instead of the raw AddArtifactsResponse.

Arguments:

  • corpus_id - The corpus to ingest the artifact into.
  • artifact_id - ID of an existing artifact to add.
  • api_key - Optional API key forwarded to :meth:add_artifact.

Returns:

IngestResult with ingestion_status.status set to "PENDING" (or whatever the server returns immediately after the add call).

Raises:

  • ProblemDetailsException - If the request fails.

CorpusResourceV1.upload_and_ingest

async def upload_and_ingest(corpus_id: str,
                            *,
                            file: Optional[IO[bytes]] = None,
                            content: Optional[bytes] = None,
                            file_name: Optional[str] = None,
                            content_type: str = "application/octet-stream",
                            api_key: Optional[str] = None) -> IngestResult

Upload a file and add it to the corpus in one call.

Exactly one of file or content must be supplied.

The method performs the full presign → S3 upload → finalize → add_artifact sequence and returns an :class:IngestResult.

Arguments:

  • corpus_id - The corpus to ingest the artifact into.
  • file - A file-like object opened in binary mode. The caller is responsible for opening and closing the file.
  • content - Raw bytes to upload.
  • file_name - File name to register with the artifact. Required when content is provided; optional (but recommended) when file is provided.
  • content_type - MIME type of the content (default "application/octet-stream").
  • api_key - Optional API key forwarded to :meth:add_artifact.

Returns:

IngestResult with the artifact and initial ingestion status.

Raises:

  • ValueError - If both or neither of file/content are provided, or if file_name is missing when content is used.
  • ProblemDetailsException - If any API call fails.

CorpusResourceV1.ingest_and_poll

async def ingest_and_poll(corpus_id: str,
                          *,
                          artifact_id: Optional[str] = None,
                          file: Optional[IO[bytes]] = None,
                          content: Optional[bytes] = None,
                          file_name: Optional[str] = None,
                          content_type: str = "application/octet-stream",
                          api_key: Optional[str] = None,
                          timeout: float = 300.0,
                          initial_interval: float = 2.0,
                          max_interval: float = 30.0,
                          backoff_factor: float = 1.5) -> IngestResult

Ingest an artifact and poll until ingestion completes (or times out).

Dispatch rules:

  • If artifact_id is supplied: calls :meth:ingest.
  • Otherwise: calls :meth:upload_and_ingest with file/content.

After the initial ingest call the method polls :meth:get_ingestion with exponential back-off until the status is PROCESSED or FAILED, or timeout seconds elapse.

Arguments:

  • corpus_id - The target corpus.
  • artifact_id - Existing artifact ID (mutually exclusive with file/content).
  • file - File-like object (binary). Forwarded to :meth:upload_and_ingest.
  • content - Raw bytes. Forwarded to :meth:upload_and_ingest.
  • file_name - File name hint. Forwarded to :meth:upload_and_ingest.
  • content_type - MIME type. Forwarded to :meth:upload_and_ingest.
  • api_key - Optional API key forwarded to the underlying ingest methods.
  • timeout - Maximum seconds to wait for ingestion to finish (default 300).
  • initial_interval - Seconds to wait before the first poll (default 2.0).
  • max_interval - Maximum seconds between polls (default 30.0).
  • backoff_factor - Multiplier applied after each poll (default 1.5).

Returns:

IngestResult reflecting the terminal ingestion status.

Raises:

  • TimeoutError - If ingestion does not complete within timeout seconds.
  • ProblemDetailsException - If any underlying API call fails.

Chat (v1)

Reference for magick_mind.resources.v1.chat.

End User (v1)

Reference for magick_mind.resources.v1.end_user.

On this page

magick_mind.resources.v1.corpusCorpusResourceV1CorpusResourceV1.createCorpusResourceV1.getCorpusResourceV1.listCorpusResourceV1.updateCorpusResourceV1.deleteCorpusResourceV1.add_artifactCorpusResourceV1.add_artifactsCorpusResourceV1.remove_artifactCorpusResourceV1.list_artifactsCorpusResourceV1.list_ingestionCorpusResourceV1.get_ingestionCorpusResourceV1.get_artifact_statusCorpusResourceV1.list_artifact_statusesCorpusResourceV1.queryCorpusResourceV1.ingestCorpusResourceV1.upload_and_ingestCorpusResourceV1.ingest_and_poll