diff --git a/examples/plugins/subtitles/main.go b/examples/plugins/subtitles/main.go
new file mode 100644
index 000000000..bce45fbc2
--- /dev/null
+++ b/examples/plugins/subtitles/main.go
@@ -0,0 +1,130 @@
+// Command subtitles is an example parser plugin: it turns SRT and WebVTT
+// subtitle files into a Markdown transcript, one paragraph per cue with its
+// start time. It also keeps a per-workspace count of parsed files through
+// the Host API key-value store, to show how a plugin keeps state without a
+// database.
+//
+// Build a package with ./package.sh.
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"log"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/Tencent/WeKnora/pluginsdk"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// Version must match plugin.yaml.
+const Version = "1.0.0"
+
+func main() {
+	p := pluginsdk.New(pluginsdk.Info{ID: "weknora-examples.subtitles", Version: Version})
+	p.Parser("subtitles", pluginsdk.ParserFunc(parse))
+	if err := p.Serve(); err != nil {
+		log.Fatal(err)
+	}
+}
+
+// timing matches a cue timing line: 00:01:02,500 --> 00:01:04,000 (SRT) or
+// 01:02.500 --> 01:04.000 (WebVTT, hours optional).
+var timing = regexp.MustCompile(`^\s*((?:\d+:)?\d{2}:\d{2}[.,]\d{3})\s*-->\s*`)
+
+// tags strips WebVTT / SRT inline markup such as <i> or <v Speaker>.
+var tags = regexp.MustCompile(`<[^>]*>`)
+
+type cue struct {
+	start string
+	text  []string
+}
+
+func parse(ctx context.Context, call *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
+	if len(in.Content) == 0 {
+		return nil, pluginapi.Errorf(pluginapi.CodeInvalidConfig, "the subtitle file is empty")
+	}
+	cues := parseCues(in.Content)
+	if len(cues) == 0 {
+		return nil, pluginapi.Errorf(
+			pluginapi.CodeInvalidConfig,
+			"no subtitle cues found; is this an .srt or .vtt file?",
+		)
+	}
+	title := strings.TrimSpace(in.Title)
+	if title == "" {
+		title = strings.TrimSuffix(in.FileName, "."+in.FileType)
+	}
+	var md strings.Builder
+	fmt.Fprintf(&md, "# %s\n\n", title)
+	for _, c := range cues {
+		fmt.Fprintf(&md, "**[%s]** %s\n\n", shortTime(c.start), strings.Join(c.text, " "))
+	}
+	countParsed(ctx, call)
+	return &pluginapi.ParseOutput{
+		Markdown: strings.TrimSpace(md.String()) + "\n",
+		Metadata: map[string]string{"cues": fmt.Sprint(len(cues))},
+	}, nil
+}
+
+func parseCues(data []byte) []cue {
+	data = bytes.TrimPrefix(data, []byte("\xef\xbb\xbf"))
+	var out []cue
+	var cur *cue
+	sc := bufio.NewScanner(bytes.NewReader(data))
+	sc.Buffer(make([]byte, 0, 64<<10), 1<<20)
+	for sc.Scan() {
+		line := strings.TrimRight(sc.Text(), "\r")
+		if m := timing.FindStringSubmatch(line); m != nil {
+			out = append(out, cue{start: m[1]})
+			cur = &out[len(out)-1]
+			continue
+		}
+		if strings.TrimSpace(line) == "" {
+			cur = nil
+			continue
+		}
+		if cur == nil {
+			continue // cue numbers, WEBVTT header, NOTE blocks
+		}
+		if text := strings.TrimSpace(tags.ReplaceAllString(line, "")); text != "" {
+			cur.text = append(cur.text, text)
+		}
+	}
+	kept := out[:0]
+	for _, c := range out {
+		if len(c.text) > 0 {
+			kept = append(kept, c)
+		}
+	}
+	return kept
+}
+
+// shortTime drops milliseconds and a zero hour: 00:01:02,500 → 01:02.
+func shortTime(t string) string {
+	t = strings.NewReplacer(",", ".").Replace(t)
+	if i := strings.IndexByte(t, '.'); i >= 0 {
+		t = t[:i]
+	}
+	return strings.TrimPrefix(t, "00:")
+}
+
+// countParsed bumps the workspace's parsed-file counter. It is best effort:
+// a missing grant or an unreachable host must not fail the parse.
+func countParsed(ctx context.Context, call *pluginsdk.Call) {
+	host := call.Host()
+	if host == nil {
+		return
+	}
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	var n int
+	if _, err := host.KVGet(ctx, "stats/parsed", &n); err != nil {
+		return
+	}
+	_ = host.KVPut(ctx, "stats/parsed", n+1, 0)
+}
diff --git a/examples/plugins/subtitles/main_test.go b/examples/plugins/subtitles/main_test.go
new file mode 100644
index 000000000..66ee0f3c9
--- /dev/null
+++ b/examples/plugins/subtitles/main_test.go
@@ -0,0 +1,146 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"go/parser"
+	"go/token"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"github.com/Tencent/WeKnora/internal/application/repository"
+	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/pluginsdk"
+	"github.com/Tencent/WeKnora/pluginsdk/client"
+	"github.com/Tencent/WeKnora/pluginsdk/conformance"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+const srt = "\ufeff1\r\n00:00:01,000 --> 00:00:03,500\r\nHello <i>world</i>.\r\n\r\n" +
+	"2\r\n00:01:02,250 --> 00:01:04,000\r\nSecond line\r\ncontinues here.\r\n"
+
+const vtt = `WEBVTT
+
+NOTE a comment
+
+01:02.500 --> 01:04.000
+<v Ada>Plugins run out of process.
+
+1:00:00.000 --> 1:00:02.000 align:start
+Bye.
+`
+
+func TestParseSRTAndVTT(t *testing.T) {
+	ctx := context.Background()
+	out, err := parse(
+		ctx,
+		&pluginsdk.Call{},
+		pluginapi.ParseInput{FileName: "talk.srt", FileType: "srt", Content: []byte(srt)},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := "# talk\n\n**[00:01]** Hello world.\n\n**[01:02]** Second line continues here.\n"
+	if out.Markdown != want || out.Metadata["cues"] != "2" {
+		t.Fatalf("srt markdown = %q", out.Markdown)
+	}
+	out, err = parse(
+		ctx,
+		&pluginsdk.Call{},
+		pluginapi.ParseInput{FileName: "t.vtt", FileType: "vtt", Title: "Talk", Content: []byte(vtt)},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(out.Markdown, "# Talk") ||
+		!strings.Contains(out.Markdown, "**[01:02]** Plugins run out of process.") ||
+		!strings.Contains(out.Markdown, "**[1:00:00]** Bye.") {
+		t.Fatalf("vtt markdown = %q", out.Markdown)
+	}
+	for _, bad := range []string{"", "just some text\nwith no cues\n"} {
+		_, err := parse(
+			ctx,
+			&pluginsdk.Call{},
+			pluginapi.ParseInput{FileName: "x.srt", FileType: "srt", Content: []byte(bad)},
+		)
+		if e, ok := pluginapi.AsError(err); !ok || e.Code != pluginapi.CodeInvalidConfig || e.Retryable {
+			t.Errorf("%q: want a permanent invalid_config error, got %v", bad, err)
+		}
+	}
+}
+
+func TestParseCountsThroughTheHostAPI(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&types.PluginKV{}); err != nil {
+		t.Fatal(err)
+	}
+	iss := hostapi.NewIssuer([]byte("k"))
+	r := gin.New()
+	hostapi.NewHandler(iss, hostapi.NewKV(repository.NewPluginKVRepository(db))).Register(r)
+	srv := httptest.NewServer(r)
+	defer srv.Close()
+
+	tok, _, _ := iss.Issue("weknora-examples.subtitles", Version, 3, []string{"kv"})
+	call := &pluginsdk.Call{
+		Context: pluginapi.Context{TenantID: 3, Host: &pluginapi.HostAccess{URL: srv.URL, Token: tok}},
+	}
+	in := pluginapi.ParseInput{FileName: "a.srt", FileType: "srt", Content: []byte(srt)}
+	for range 2 {
+		if _, err := parse(context.Background(), call, in); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var n int
+	if ok, err := call.Host().KVGet(context.Background(), "stats/parsed", &n); !ok || err != nil || n != 2 {
+		t.Fatalf("parsed count = %d (%v, %v)", n, ok, err)
+	}
+}
+
+func TestConformsAndStaysExternal(t *testing.T) {
+	p := pluginsdk.New(pluginsdk.Info{ID: "weknora-examples.subtitles", Version: Version})
+	p.Parser("subtitles", pluginsdk.ParserFunc(parse))
+	srv := httptest.NewServer(p.Handler())
+	defer srv.Close()
+	rep := conformance.Run(context.Background(), conformance.Target{
+		Client: client.New(srv.URL, nil, nil),
+		Raw: func(ctx context.Context, path string, body []byte) (*http.Response, error) {
+			req, _ := http.NewRequestWithContext(ctx, http.MethodPost, srv.URL+path, bytes.NewReader(body))
+			return http.DefaultClient.Do(req)
+		},
+	})
+	for _, res := range rep.Results {
+		if !res.Passed {
+			t.Errorf("%s: %s", res.Name, res.Detail)
+		}
+	}
+	files, _ := filepath.Glob("*.go")
+	for _, f := range files {
+		if strings.HasSuffix(f, "_test.go") {
+			continue
+		}
+		src, _ := os.ReadFile(f)
+		parsed, err := parser.ParseFile(token.NewFileSet(), f, src, parser.ImportsOnly)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, imp := range parsed.Imports {
+			if strings.Contains(imp.Path.Value, "WeKnora/internal") {
+				t.Errorf("%s imports %s", f, imp.Path.Value)
+			}
+		}
+	}
+}
diff --git a/examples/plugins/subtitles/package.sh b/examples/plugins/subtitles/package.sh
new file mode 100755
index 000000000..09f3d236e
--- /dev/null
+++ b/examples/plugins/subtitles/package.sh
@@ -0,0 +1,20 @@
+#!/usr/bin/env bash
+# Builds the subtitles example plugin for the platforms WeKnora runs on and zips it
+# into a .wkp package: plugin.yaml and bin/<os>-<arch>/subtitles.
+set -euo pipefail
+cd "$(dirname "$0")"
+version=$(sed -n 's/^version: //p' plugin.yaml)
+out=${1:-"weknora-examples-subtitles-${version}.wkp"}
+stage=$(mktemp -d)
+trap 'rm -rf "$stage"' EXIT
+cp plugin.yaml "$stage"/
+for target in linux/amd64 linux/arm64 darwin/amd64 darwin/arm64 windows/amd64; do
+  os=${target%/*}
+  arch=${target#*/}
+  bin="$stage/bin/${os}-${arch}/subtitles"
+  [ "$os" = windows ] && bin="$bin.exe"
+  CGO_ENABLED=0 GOOS=$os GOARCH=$arch go build -trimpath -ldflags="-s -w" -o "$bin" .
+done
+rm -f "$out"
+(cd "$stage" && zip -qr - .) > "$out"
+echo "wrote $out"
diff --git a/examples/plugins/subtitles/plugin.yaml b/examples/plugins/subtitles/plugin.yaml
new file mode 100644
index 000000000..86252e3fc
--- /dev/null
+++ b/examples/plugins/subtitles/plugin.yaml
@@ -0,0 +1,23 @@
+schemaVersion: 1
+id: weknora-examples.subtitles
+version: 1.0.0
+apiVersion: weknora.plugin/v1
+name: { en-US: Subtitles, zh-CN: 字幕解析 }
+description:
+  en-US: Parse SRT and WebVTT subtitles into a searchable transcript.
+  zh-CN: 把 SRT、WebVTT 字幕解析为可检索的文字稿。
+publisher: { id: weknora-examples, name: WeKnora examples }
+homepage: https://github.com/Tencent/WeKnora/tree/main/examples/plugins/subtitles
+license: MIT
+runtime: { type: host, kind: binary, entry: "bin/{os}-{arch}/subtitles" }
+permissions:
+  # Counts parsed files per workspace in the Host API key-value store.
+  hostApi: [kv]
+contributes:
+  parsers:
+    - id: subtitles
+      name: { en-US: Subtitles, zh-CN: 字幕解析 }
+      description:
+        en-US: SRT and WebVTT subtitles as a transcript with cue times
+        zh-CN: 带时间点的字幕文字稿
+      fileTypes: [srt, vtt]
diff --git a/frontend/src/api/system/index.ts b/frontend/src/api/system/index.ts
index fa717354b..978564371 100644
--- a/frontend/src/api/system/index.ts
+++ b/frontend/src/api/system/index.ts
@@ -107,6 +107,10 @@ export interface ParserEngineInfo {
   FileTypes: string[]
   Available?: boolean
   UnavailableReason?: string
+  /** Set on engines an installed plugin provides (Name is then acme.ocr/ocr). */
+  PluginID?: string
+  /** Localized names of plugin engines, keyed by locale; "default" as fallback. */
+  DisplayNames?: Record<string, string>
 }
 
 /** 解析引擎配置（引擎连接参数存空间；聊天附件解析策略在智能体中配置） */
diff --git a/frontend/src/views/knowledge/settings/KBParserSettings.vue b/frontend/src/views/knowledge/settings/KBParserSettings.vue
index 107eb2dbd..7332b3dfd 100644
--- a/frontend/src/views/knowledge/settings/KBParserSettings.vue
+++ b/frontend/src/views/knowledge/settings/KBParserSettings.vue
@@ -66,6 +66,7 @@
 </template>
 
 <script setup lang="ts">
+import { pickLocale } from '@/utils/localizedText'
 import { ref, watch, computed, onMounted, onUnmounted } from 'vue'
 import { useI18n } from 'vue-i18n'
 import { type ParserEngineInfo } from '@/api/system'
@@ -73,10 +74,13 @@ import { useEditorResourcesStore } from '@/stores/editorResources'
 import { useUIStore } from '@/stores/ui'
 import { storeToRefs } from 'pinia'
 
-const { t } = useI18n()
+const { t, locale } = useI18n()
 const editorResources = useEditorResourcesStore()
 
 function getEngineDisplayName(engineName: string): string {
+  // Plugin engines name themselves.
+  const names = parserEngines.value.find((e) => e.Name === engineName)?.DisplayNames
+  if (names) return pickLocale(names, locale.value) || names.default || engineName
   const key = `kbSettings.parser.engines.${engineName}.name`
   const translated = t(key)
   return translated !== key ? translated : engineName
diff --git a/frontend/src/views/settings/ParserEngineSettings.vue b/frontend/src/views/settings/ParserEngineSettings.vue
index 436d255ee..e046eb566 100644
--- a/frontend/src/views/settings/ParserEngineSettings.vue
+++ b/frontend/src/views/settings/ParserEngineSettings.vue
@@ -409,6 +409,7 @@
 </template>
 
 <script setup lang="ts">
+import { pickLocale } from '@/utils/localizedText'
 import { ref, computed, onMounted, nextTick } from 'vue'
 import { useI18n } from 'vue-i18n'
 import { useUIStore } from '@/stores/ui'
@@ -425,7 +426,7 @@ import {
 } from '@/api/system'
 import { getWeKnoraCloudStatus } from '@/api/model'
 
-const { t } = useI18n()
+const { t, locale } = useI18n()
 const uiStore = useUIStore()
 const authStore = useAuthStore()
 
@@ -546,6 +547,9 @@ function engineInitial(engineName: string): string {
 }
 
 function getEngineDisplayName(engineName: string): string {
+  // Plugin engines name themselves.
+  const names = engines.value.find((e) => e.Name === engineName)?.DisplayNames
+  if (names) return pickLocale(names, locale.value) || names.default || engineName
   const key = `kbSettings.parser.engines.${engineName}.name`
   const translated = t(key)
   return translated !== key ? translated : engineName
diff --git a/internal/application/repository/plugin_kv.go b/internal/application/repository/plugin_kv.go
new file mode 100644
index 000000000..064d72734
--- /dev/null
+++ b/internal/application/repository/plugin_kv.go
@@ -0,0 +1,90 @@
+package repository
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/internal/types/interfaces"
+)
+
+// pluginKVRepository stores plugins' Host API key-value data (plugin_kv,
+// migration 000117).
+type pluginKVRepository struct {
+	db *gorm.DB
+}
+
+// NewPluginKVRepository wires the repository into the container.
+func NewPluginKVRepository(db *gorm.DB) interfaces.PluginKVRepository {
+	return &pluginKVRepository{db: db}
+}
+
+// live restricts a query to entries that have not expired.
+func live(q *gorm.DB, now time.Time) *gorm.DB {
+	return q.Where("expires_at IS NULL OR expires_at > ?", now)
+}
+
+func (r *pluginKVRepository) scope(ctx context.Context, pluginID string, tenantID uint64) *gorm.DB {
+	return r.db.WithContext(ctx).Model(&types.PluginKV{}).
+		Where("plugin_id = ? AND tenant_id = ?", pluginID, tenantID)
+}
+
+func (r *pluginKVRepository) Get(
+	ctx context.Context, pluginID string, tenantID uint64, key string,
+) (*types.PluginKV, error) {
+	var rows []types.PluginKV
+	err := live(r.scope(ctx, pluginID, tenantID).Where("key = ?", key), time.Now()).Limit(1).Find(&rows).Error
+	if err != nil || len(rows) == 0 {
+		return nil, err
+	}
+	return &rows[0], nil
+}
+
+func (r *pluginKVRepository) Put(ctx context.Context, e *types.PluginKV) error {
+	return r.db.WithContext(ctx).Clauses(clause.OnConflict{
+		Columns:   []clause.Column{{Name: "plugin_id"}, {Name: "tenant_id"}, {Name: "key"}},
+		DoUpdates: clause.AssignmentColumns([]string{"value", "expires_at", "updated_at"}),
+	}).Create(e).Error
+}
+
+func (r *pluginKVRepository) Delete(ctx context.Context, pluginID string, tenantID uint64, key string) (bool, error) {
+	res := r.db.WithContext(ctx).
+		Where("plugin_id = ? AND tenant_id = ? AND key = ?", pluginID, tenantID, key).
+		Delete(&types.PluginKV{})
+	return res.RowsAffected > 0, res.Error
+}
+
+// escapeLike makes a prefix match literally in LIKE.
+func escapeLike(s string) string {
+	return strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(s)
+}
+
+func (r *pluginKVRepository) List(
+	ctx context.Context, pluginID string, tenantID uint64, prefix, after string, limit int,
+) ([]types.PluginKV, error) {
+	q := live(r.scope(ctx, pluginID, tenantID), time.Now())
+	if prefix != "" {
+		q = q.Where(`key LIKE ? ESCAPE '\'`, escapeLike(prefix)+"%")
+	}
+	if after != "" {
+		q = q.Where("key > ?", after)
+	}
+	var rows []types.PluginKV
+	err := q.Order("key ASC").Limit(limit).Find(&rows).Error
+	return rows, err
+}
+
+func (r *pluginKVRepository) Count(ctx context.Context, pluginID string, tenantID uint64) (int64, error) {
+	var n int64
+	err := live(r.scope(ctx, pluginID, tenantID), time.Now()).Count(&n).Error
+	return n, err
+}
+
+func (r *pluginKVRepository) DeleteExpired(ctx context.Context, now time.Time) (int64, error) {
+	res := r.db.WithContext(ctx).Where("expires_at IS NOT NULL AND expires_at <= ?", now).Delete(&types.PluginKV{})
+	return res.RowsAffected, res.Error
+}
diff --git a/internal/application/repository/plugin_test.go b/internal/application/repository/plugin_test.go
index b603c4700..8d945fa4a 100644
--- a/internal/application/repository/plugin_test.go
+++ b/internal/application/repository/plugin_test.go
@@ -93,3 +93,52 @@ func TestPluginTenantSettingUpsertColumns(t *testing.T) {
 	require.False(t, got.Enabled)
 	require.JSONEq(t, `{"region":"eu"}`, string(got.Config), "flipping the switch must keep the config")
 }
+
+func TestPluginKVRepository(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
+	require.NoError(t, err)
+	require.NoError(t, db.AutoMigrate(&types.PluginKV{}))
+	repo := NewPluginKVRepository(db)
+	ctx := context.Background()
+	put := func(tenant uint64, key, value string, expires *time.Time) {
+		t.Helper()
+		require.NoError(t, repo.Put(ctx, &types.PluginKV{
+			PluginID: "acme.x", TenantID: tenant, Key: key, Value: types.JSON(value),
+			ExpiresAt: expires, UpdatedAt: time.Now(),
+		}))
+	}
+	past := time.Now().Add(-time.Minute)
+	put(1, "cursor:a", `1`, nil)
+	put(1, "cursor:b", `2`, nil)
+	put(1, "cursor_x", `3`, nil) // "_" must not act as a LIKE wildcard
+	put(1, "old", `4`, &past)
+	put(2, "cursor:a", `99`, nil)
+	put(1, "cursor:a", `10`, nil) // upsert
+
+	got, err := repo.Get(ctx, "acme.x", 1, "cursor:a")
+	require.NoError(t, err)
+	require.JSONEq(t, `10`, string(got.Value))
+	got, err = repo.Get(ctx, "acme.x", 1, "old")
+	require.NoError(t, err)
+	require.Nil(t, got, "expired entries read as missing")
+
+	list, err := repo.List(ctx, "acme.x", 1, "cursor:", "", 10)
+	require.NoError(t, err)
+	require.Len(t, list, 2)
+	list, err = repo.List(ctx, "acme.x", 1, "cursor:", "cursor:a", 10)
+	require.NoError(t, err)
+	require.Len(t, list, 1)
+	require.Equal(t, "cursor:b", list[0].Key)
+
+	n, err := repo.Count(ctx, "acme.x", 1)
+	require.NoError(t, err)
+	require.EqualValues(t, 3, n)
+	removed, err := repo.DeleteExpired(ctx, time.Now())
+	require.NoError(t, err)
+	require.EqualValues(t, 1, removed)
+	ok, err := repo.Delete(ctx, "acme.x", 1, "cursor:b")
+	require.NoError(t, err)
+	require.True(t, ok)
+	got, _ = repo.Get(ctx, "acme.x", 2, "cursor:a")
+	require.JSONEq(t, `99`, string(got.Value), "tenants are separate partitions")
+}
diff --git a/internal/application/service/knowledge_util.go b/internal/application/service/knowledge_util.go
index e520edbc6..1706a5be3 100644
--- a/internal/application/service/knowledge_util.go
+++ b/internal/application/service/knowledge_util.go
@@ -10,6 +10,7 @@ import (
 	"mime/multipart"
 	"net/http"
 	"os"
+	"slices"
 	"strings"
 	"time"
 
@@ -56,8 +57,11 @@ func isSupportedImportExtension(ext string) bool {
 	if ext == "" || ext == unknownFileType {
 		return false
 	}
-	_, ok := supportedImportFileExtensions[ext]
-	return ok
+	if _, ok := supportedImportFileExtensions[ext]; ok {
+		return true
+	}
+	// Installed plugins' parser engines bring their own file types.
+	return slices.Contains(docparser.PluginFileTypes(), ext)
 }
 
 // isValidFileType checks if a filename's extension is supported for import.
diff --git a/internal/container/container.go b/internal/container/container.go
index 2ea8cea6f..2c940b1f6 100644
--- a/internal/container/container.go
+++ b/internal/container/container.go
@@ -169,8 +169,11 @@ func BuildContainer(container *dig.Container) *dig.Container {
 	must(container.Provide(activate.NewModelVendors))
 	must(container.Provide(pluginhost.NewManager))
 	must(container.Provide(newPluginInvoker))
+	must(container.Provide(repository.NewPluginKVRepository))
+	must(container.Provide(newPluginHostAPI))
 	must(container.Provide(activate.NewWebSearch))
 	must(container.Provide(activate.NewConnectors))
+	must(container.Provide(activate.NewParsers))
 	must(container.Provide(newMCPServiceRepository))
 	must(container.Provide(repository.NewMCPToolApprovalRepository))
 	must(container.Provide(repository.NewMCPOAuthRepository))
@@ -1854,8 +1857,10 @@ func installPluginGate(
 	webSearch *handler.WebSearchProviderHandler,
 	dataSources *handler.DataSourceHandler,
 	imHandler *handler.IMHandler,
+	system *handler.SystemHandler,
 ) {
-	for _, h := range []interface{ SetPluginGate(interfaces.PluginGate) }{models, webSearch, dataSources, imHandler} {
+	gated := []interface{ SetPluginGate(interfaces.PluginGate) }{models, webSearch, dataSources, imHandler, system}
+	for _, h := range gated {
 		h.SetPluginGate(gate)
 	}
 }
diff --git a/internal/container/plugin_runtime.go b/internal/container/plugin_runtime.go
index fc84f88e1..d6a8bf947 100644
--- a/internal/container/plugin_runtime.go
+++ b/internal/container/plugin_runtime.go
@@ -2,6 +2,10 @@ package container
 
 import (
 	"context"
+	"fmt"
+	"os"
+	"strings"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/dig"
@@ -13,6 +17,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/handler"
 	"github.com/Tencent/WeKnora/internal/plugin/activate"
 	"github.com/Tencent/WeKnora/internal/plugin/host"
+	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
 	"github.com/Tencent/WeKnora/internal/plugin/install"
 	"github.com/Tencent/WeKnora/internal/plugin/reconcile"
 	pluginregistry "github.com/Tencent/WeKnora/internal/plugin/registry"
@@ -38,6 +43,7 @@ type pluginActivators struct {
 	Host       *host.Manager
 	WebSearch  *activate.WebSearch
 	Connectors *activate.Connectors
+	Parsers    *activate.Parsers
 	MCP        *activate.MCPServers
 	Skills     *activate.Skills
 	Vendors    *activate.ModelVendors
@@ -47,7 +53,29 @@ type pluginActivators struct {
 // list orders the activators: the host first, so a code plugin's process is
 // running before anything routes calls to it.
 func (a pluginActivators) list() []reconcile.Activator {
-	return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Vendors, a.MCP, a.Skills}
+	return []reconcile.Activator{a.Host, a.WebSearch, a.Connectors, a.Parsers, a.Vendors, a.MCP, a.Skills}
+}
+
+// newPluginHostAPI serves the Host API and gives calls a way back to it: the
+// embedded host's plugins reach this node on loopback.
+func newPluginHostAPI(
+	cfg *config.Config, iv *activate.Invoker, repo interfaces.PluginKVRepository, cleaner interfaces.ResourceCleaner,
+) *hostapi.Handler {
+	issuer := hostapi.NewIssuerFromEnv()
+	url := strings.TrimSpace(os.Getenv("WEKNORA_PLUGIN_HOST_API_URL"))
+	if url == "" {
+		port := 8080
+		if cfg != nil && cfg.Server != nil && cfg.Server.Port > 0 {
+			port = cfg.Server.Port
+		}
+		url = fmt.Sprintf("http://127.0.0.1:%d", port)
+	}
+	iv.SetHostAPI(issuer, url)
+	kv := hostapi.NewKV(repo)
+	ctx, cancel := context.WithCancel(context.Background())
+	kv.StartSweeper(ctx, 10*time.Minute)
+	cleaner.RegisterWithName("PluginKVSweeper", func() error { cancel(); return nil })
+	return hostapi.NewHandler(issuer, kv)
 }
 
 // bindPluginActivators hands the activators what they need once the plugin
diff --git a/internal/database/migration_sqlite_versioned_schema_test.go b/internal/database/migration_sqlite_versioned_schema_test.go
index f87a3a806..9ff7076e3 100644
--- a/internal/database/migration_sqlite_versioned_schema_test.go
+++ b/internal/database/migration_sqlite_versioned_schema_test.go
@@ -38,6 +38,7 @@ var versionedSQLiteTables = []string{
 	"plugin_tenant_settings",
 	"plugins",
 	"plugin_versions",
+	"plugin_kv",
 }
 
 // versionedSQLiteColumns maps each existing table to the columns that the
@@ -72,7 +73,7 @@ var versionedSQLiteColumns = map[string][]string{
 	}, // 000028
 }
 
-const expectedSQLiteMigrationVersion = 35
+const expectedSQLiteMigrationVersion = 36
 
 func TestSQLiteMigrationsCreateVersionedSchema(t *testing.T) {
 	repoRoot := sqliteRepoRoot(t)
diff --git a/internal/handler/system.go b/internal/handler/system.go
index 40cabdc53..a5e1bae46 100644
--- a/internal/handler/system.go
+++ b/internal/handler/system.go
@@ -24,6 +24,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
 	"github.com/Tencent/WeKnora/internal/logger"
 	modellimiter "github.com/Tencent/WeKnora/internal/models/limiter"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
 	"github.com/Tencent/WeKnora/internal/runtime"
 	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
@@ -38,6 +39,7 @@ type runtimeKnowledgeCanceller interface {
 
 // SystemHandler handles system-related requests
 type SystemHandler struct {
+	pluginGated
 	cfg              *config.Config
 	neo4jDriver      neo4j.Driver
 	documentReader   interfaces.DocumentReader
@@ -420,7 +422,18 @@ func (h *SystemHandler) ListParserEngines(c *gin.Context) {
 	connected := reader != nil && reader.IsConnected()
 	remoteEngines := h.fetchRemoteEngines(c.Request.Context(), reader, overrides)
 	engines := docparser.ListAllEngines(connected, overrides, remoteEngines)
-	c.JSON(200, gin.H{"code": 0, "msg": "success", "data": engines, "docreader_addr": docreaderAddr, "docreader_transport": docreaderTransport, "connected": connected})
+	// Engines of plugins the workspace has turned off leave the list.
+	enabled := h.pluginFilter(c)
+	visible := engines[:0]
+	for _, e := range engines {
+		if enabled(manifest.PointParsers, e.Name) {
+			visible = append(visible, e)
+		}
+	}
+	c.JSON(200, gin.H{
+		"code": 0, "msg": "success", "data": visible,
+		"docreader_addr": docreaderAddr, "docreader_transport": docreaderTransport, "connected": connected,
+	})
 }
 
 // ReconnectDocReader reconnects the document converter to a new (or same) DocReader address.
diff --git a/internal/infrastructure/docparser/engine_registry.go b/internal/infrastructure/docparser/engine_registry.go
index 725d7b8bb..f378fd000 100644
--- a/internal/infrastructure/docparser/engine_registry.go
+++ b/internal/infrastructure/docparser/engine_registry.go
@@ -3,6 +3,9 @@ package docparser
 import (
 	"context"
 	"fmt"
+	"sort"
+	"strings"
+	"sync"
 
 	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
@@ -41,32 +44,102 @@ type ReaderDeps struct {
 	WeKnoraCloudCredentials func(ctx context.Context) *types.WeKnoraCloudCredentials
 }
 
+// enginesMu guards the registries: plugin engines come and go while the
+// server runs.
+var enginesMu sync.RWMutex
+
 // localEngines holds all locally registered parser engines, in registration
 // order — which is also the order the engine list is shown in.
 var localEngines []EngineRegistration
 
+// pluginEngines holds engines installed plugins provide, by qualified name
+// ("acme.ocr/ocr"). They are kept apart from localEngines so the builtin
+// plugin catalog never describes them as builtins.
+var pluginEngines = map[string]EngineRegistration{}
+
+// PluginEngineInfo is what a plugin engine adds to the engine list.
+type PluginEngineInfo interface {
+	PluginID() string
+	// DisplayNames localizes the engine's name, keyed by locale.
+	DisplayNames() map[string]string
+}
+
 // RegisterEngine adds an engine to the local registry. Called from init().
 func RegisterEngine(e EngineRegistration) {
+	enginesMu.Lock()
+	defer enginesMu.Unlock()
 	localEngines = append(localEngines, e)
 }
 
-// Engines returns the locally registered engines in registration order.
+// RegisterPluginEngine adds or replaces a plugin's engine. It refuses a name
+// a builtin engine uses.
+func RegisterPluginEngine(e EngineRegistration) error {
+	enginesMu.Lock()
+	defer enginesMu.Unlock()
+	for _, builtin := range localEngines {
+		if builtin.Name() == e.Name() {
+			return fmt.Errorf("parser engine %s already exists", e.Name())
+		}
+	}
+	pluginEngines[e.Name()] = e
+	return nil
+}
+
+// UnregisterPluginEngine removes a plugin's engine.
+func UnregisterPluginEngine(name string) {
+	enginesMu.Lock()
+	defer enginesMu.Unlock()
+	delete(pluginEngines, name)
+}
+
+// Engines returns the builtin engines in registration order.
 func Engines() []EngineRegistration {
+	enginesMu.RLock()
+	defer enginesMu.RUnlock()
 	out := make([]EngineRegistration, len(localEngines))
 	copy(out, localEngines)
 	return out
 }
 
-// lookupEngine returns the locally registered engine with this name.
+// pluginEngineList returns the plugin engines sorted by name.
+func pluginEngineList() []EngineRegistration {
+	enginesMu.RLock()
+	defer enginesMu.RUnlock()
+	out := make([]EngineRegistration, 0, len(pluginEngines))
+	for _, e := range pluginEngines {
+		out = append(out, e)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Name() < out[j].Name() })
+	return out
+}
+
+// PluginFileTypes returns the file types plugin engines parse, so uploads
+// of those types are accepted.
+func PluginFileTypes() []string {
+	var out []string
+	for _, e := range pluginEngineList() {
+		out = append(out, e.FileTypes(true)...)
+	}
+	return out
+}
+
+// lookupEngine returns the registered engine with this name.
 func lookupEngine(name string) (EngineRegistration, bool) {
+	enginesMu.RLock()
+	defer enginesMu.RUnlock()
 	for _, engine := range localEngines {
 		if engine.Name() == name {
 			return engine, true
 		}
 	}
-	return nil, false
+	e, ok := pluginEngines[name]
+	return e, ok
 }
 
+// isPluginEngineName reports whether a name is a qualified plugin
+// contribution ID, which no builtin or docreader engine uses.
+func isPluginEngineName(name string) bool { return strings.Contains(name, "/") }
+
 // NewReader builds the reader for an engine.
 //
 // An empty engine name means "no explicit choice": simple formats are handled
@@ -79,6 +152,11 @@ func NewReader(
 	if registration, ok := lookupEngine(engine); ok {
 		return registration.NewReader(ctx, deps)
 	}
+	if isPluginEngineName(engine) {
+		// Its plugin was uninstalled or disabled: the Python docreader does
+		// not know it either, so say so instead of sending it there.
+		return nil, errEngineUnavailable(engine, "its plugin is not installed or not running")
+	}
 	if engine == "" && !isURL && IsSimpleFormat(fileType) {
 		return &SimpleFormatReader{}, nil
 	}
@@ -112,10 +190,11 @@ func ListAllEngines(
 		remoteMap[re.Name] = re
 	}
 
-	seen := make(map[string]bool, len(localEngines))
-	result := make([]types.ParserEngineInfo, 0, len(localEngines)+len(remoteEngines))
+	locals := Engines()
+	seen := make(map[string]bool, len(locals))
+	result := make([]types.ParserEngineInfo, 0, len(locals)+len(remoteEngines))
 
-	for _, e := range localEngines {
+	for _, e := range locals {
 		name := e.Name()
 		seen[name] = true
 
@@ -148,6 +227,18 @@ func ListAllEngines(
 		result = append(result, re)
 	}
 
+	for _, e := range pluginEngineList() {
+		available, reason := e.CheckAvailable(docreaderConnected, overrides)
+		info := types.ParserEngineInfo{
+			Name: e.Name(), Description: e.Description(), FileTypes: e.FileTypes(docreaderConnected),
+			Available: available, UnavailableReason: reason,
+		}
+		if meta, ok := e.(PluginEngineInfo); ok {
+			info.PluginID, info.DisplayNames = meta.PluginID(), meta.DisplayNames()
+		}
+		result = append(result, info)
+	}
+
 	return result
 }
 
diff --git a/internal/plugin/activate/invoke.go b/internal/plugin/activate/invoke.go
index 82617aa97..46007ac77 100644
--- a/internal/plugin/activate/invoke.go
+++ b/internal/plugin/activate/invoke.go
@@ -28,6 +28,22 @@ type Invoker struct {
 	mu      sync.RWMutex
 	tenancy *tenancy.Service
 	plugins interfaces.PluginRepository
+	// tokens and hostURL give plugins granted Host API scopes a way back.
+	tokens  TokenIssuer
+	hostURL string
+}
+
+// TokenIssuer signs the Host API token of one call.
+type TokenIssuer interface {
+	Issue(pluginID, version string, tenantID uint64, scopes []string) (string, time.Time, error)
+}
+
+// SetHostAPI lets calls carry Host API access: where plugins reach WeKnora
+// and how their tokens are signed.
+func (iv *Invoker) SetHostAPI(tokens TokenIssuer, url string) {
+	iv.mu.Lock()
+	iv.tokens, iv.hostURL = tokens, url
+	iv.mu.Unlock()
 }
 
 // NewInvoker creates an Invoker; Bind completes it.
@@ -77,8 +93,15 @@ func (iv *Invoker) Envelope(
 		env.Context.Deadline = &d
 	}
 	iv.mu.RLock()
-	t, plugins := iv.tenancy, iv.plugins
+	t, plugins, tokens, hostURL := iv.tenancy, iv.plugins, iv.tokens, iv.hostURL
 	iv.mu.RUnlock()
+	if tokens != nil && hostURL != "" && env.Context.TenantID != 0 && len(m.Permissions.HostAPI) > 0 {
+		token, _, err := tokens.Issue(m.ID, m.Version, env.Context.TenantID, m.Permissions.HostAPI)
+		if err != nil {
+			return env, err
+		}
+		env.Context.Host = &pluginapi.HostAccess{URL: hostURL, Token: token}
+	}
 	if plugins != nil && len(m.Config.SystemSchema) > 0 {
 		sys, _, err := install.OpenSystemConfig(ctx, plugins, m)
 		if err != nil {
diff --git a/internal/plugin/activate/invoke_test.go b/internal/plugin/activate/invoke_test.go
new file mode 100644
index 000000000..c7231ff6c
--- /dev/null
+++ b/internal/plugin/activate/invoke_test.go
@@ -0,0 +1,37 @@
+package activate
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/types"
+)
+
+func TestEnvelopeCarriesHostAccessOnlyForGrantedPlugins(t *testing.T) {
+	iss := hostapi.NewIssuer([]byte("k"))
+	iv := NewInvoker(fakeClients{})
+	iv.SetHostAPI(iss, "http://127.0.0.1:9")
+	ctx := context.WithValue(context.Background(), types.TenantIDContextKey, uint64(7))
+
+	granted := &manifest.Manifest{
+		ID: "acme.x", Version: "1.0.0", Permissions: manifest.Permissions{HostAPI: []string{"kv"}},
+	}
+	env, err := iv.Envelope(ctx, granted, nil)
+	if err != nil || env.Context.Host == nil || env.Context.Host.URL != "http://127.0.0.1:9" {
+		t.Fatalf("envelope = %+v, %v", env.Context, err)
+	}
+	claims, err := iss.Verify(env.Context.Host.Token)
+	if err != nil || claims.PluginID != "acme.x" || claims.TenantID != 7 || !claims.Has("kv") {
+		t.Fatalf("token claims = %+v, %v", claims, err)
+	}
+
+	plain := &manifest.Manifest{ID: "acme.y", Version: "1.0.0"}
+	if env, _ := iv.Envelope(ctx, plain, nil); env.Context.Host != nil {
+		t.Fatal("a plugin without Host API scopes must get no token")
+	}
+	if env, _ := iv.Envelope(context.Background(), granted, nil); env.Context.Host != nil {
+		t.Fatal("a call without a tenant must get no token")
+	}
+}
diff --git a/internal/plugin/activate/parsers.go b/internal/plugin/activate/parsers.go
new file mode 100644
index 000000000..59cd4d0cd
--- /dev/null
+++ b/internal/plugin/activate/parsers.go
@@ -0,0 +1,152 @@
+package activate
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"path"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
+	"github.com/Tencent/WeKnora/internal/plugin/manifest"
+	"github.com/Tencent/WeKnora/internal/plugin/reconcile"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/internal/types/interfaces"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// parseTimeout bounds one plugin parse when the caller set no deadline.
+const parseTimeout = 10 * time.Minute
+
+// Parsers registers code plugins' document parsers as parser engines under
+// their qualified IDs, next to the builtin engines and the docreader's.
+type Parsers struct {
+	iv *Invoker
+
+	mu         sync.Mutex
+	registered map[string][]string
+}
+
+// NewParsers creates the parser activator.
+func NewParsers(iv *Invoker) *Parsers {
+	return &Parsers{iv: iv, registered: map[string][]string{}}
+}
+
+// Name implements reconcile.Activator.
+func (a *Parsers) Name() string { return "parsers" }
+
+// Activate implements reconcile.Activator: all of a plugin's parsers or none.
+func (a *Parsers) Activate(_ context.Context, l *reconcile.Loaded) error {
+	var ids []string
+	for _, c := range l.Manifest.Contributes[manifest.PointParsers] {
+		e := &pluginEngine{
+			iv: a.iv, m: l.Manifest, local: c.ID, name: manifest.QualifiedID(l.Manifest.ID, c.ID),
+			description: c.Description.Default, names: displayNames(c.Name), fileTypes: c.FileTypes,
+		}
+		if err := docparser.RegisterPluginEngine(e); err != nil {
+			for _, done := range ids {
+				docparser.UnregisterPluginEngine(done)
+			}
+			return fmt.Errorf("parser %s: %w", c.ID, err)
+		}
+		ids = append(ids, e.name)
+	}
+	a.mu.Lock()
+	a.registered[l.Manifest.ID] = ids
+	a.mu.Unlock()
+	return nil
+}
+
+// Deactivate implements reconcile.Activator.
+func (a *Parsers) Deactivate(_ context.Context, pluginID string) error {
+	a.mu.Lock()
+	ids := a.registered[pluginID]
+	delete(a.registered, pluginID)
+	a.mu.Unlock()
+	for _, id := range ids {
+		docparser.UnregisterPluginEngine(id)
+	}
+	return nil
+}
+
+func displayNames(t manifest.LocalizedText) map[string]string {
+	out := map[string]string{}
+	for k, v := range t.Locales {
+		out[k] = v
+	}
+	if t.Default != "" {
+		out["default"] = t.Default
+	}
+	return out
+}
+
+// pluginEngine is a plugin parser as a docparser engine.
+type pluginEngine struct {
+	iv          *Invoker
+	m           *manifest.Manifest
+	local, name string
+	description string
+	names       map[string]string
+	fileTypes   []string
+}
+
+var (
+	_ docparser.EngineRegistration = (*pluginEngine)(nil)
+	_ docparser.PluginEngineInfo   = (*pluginEngine)(nil)
+)
+
+func (e *pluginEngine) Name() string                    { return e.name }
+func (e *pluginEngine) Description() string             { return e.description }
+func (e *pluginEngine) FileTypes(bool) []string         { return e.fileTypes }
+func (e *pluginEngine) PluginID() string                { return e.m.ID }
+func (e *pluginEngine) DisplayNames() map[string]string { return e.names }
+func (e *pluginEngine) CheckAvailable(bool, map[string]string) (bool, string) {
+	if _, err := e.iv.clients.Client(e.m.ID); err != nil {
+		return false, err.Error()
+	}
+	return true, ""
+}
+
+// NewReader implements docparser.EngineRegistration. deps.Overrides carry
+// other engines' credentials and are deliberately not passed on; the
+// plugin's own settings travel in its system and tenant configuration.
+func (e *pluginEngine) NewReader(context.Context, docparser.ReaderDeps) (interfaces.DocReader, error) {
+	return &remoteParser{engine: e}, nil
+}
+
+// remoteParser reads one document through the plugin.
+type remoteParser struct{ engine *pluginEngine }
+
+// Read implements interfaces.DocReader. A retryable plugin failure (the
+// plugin is down, rate limited) is returned as an error so the task is
+// retried; any other failure is final and reported in the result.
+func (r *remoteParser) Read(ctx context.Context, req *types.ReadRequest) (*types.ReadResult, error) {
+	ctx, cancel := withDefaultTimeout(ctx, parseTimeout)
+	defer cancel()
+	in := pluginapi.ParseInput{
+		FileName: req.FileName, FileType: strings.ToLower(strings.TrimPrefix(req.FileType, ".")),
+		Content: req.FileContent, URL: req.URL, Title: req.Title,
+	}
+	var out pluginapi.ParseOutput
+	err := r.engine.iv.Call(ctx, r.engine.m, pluginapi.ParsePath(r.engine.local), nil, in, &out)
+	if err != nil {
+		var pe *pluginapi.Error
+		if errors.As(err, &pe) && !pe.Retryable {
+			return &types.ReadResult{Error: err.Error()}, nil
+		}
+		return nil, err
+	}
+	res := &types.ReadResult{MarkdownContent: out.Markdown, Metadata: out.Metadata}
+	for _, img := range out.Images {
+		if img.OriginalRef == "" || len(img.Data) == 0 {
+			continue
+		}
+		res.ImageRefs = append(res.ImageRefs, types.ImageRef{
+			Filename: path.Base(img.OriginalRef), OriginalRef: img.OriginalRef,
+			MimeType: img.MimeType, ImageData: img.Data,
+		})
+	}
+	return res, nil
+}
diff --git a/internal/plugin/activate/parsers_test.go b/internal/plugin/activate/parsers_test.go
new file mode 100644
index 000000000..5468f9be5
--- /dev/null
+++ b/internal/plugin/activate/parsers_test.go
@@ -0,0 +1,120 @@
+package activate
+
+import (
+	"context"
+	"net/http/httptest"
+	"slices"
+	"strings"
+	"testing"
+
+	"github.com/Tencent/WeKnora/internal/infrastructure/docparser"
+	"github.com/Tencent/WeKnora/internal/plugin/pkg"
+	"github.com/Tencent/WeKnora/internal/plugin/plugintest"
+	"github.com/Tencent/WeKnora/internal/plugin/reconcile"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/pluginsdk"
+	"github.com/Tencent/WeKnora/pluginsdk/client"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+const ocrManifest = `schemaVersion: 1
+id: acme.ocr
+version: 1.0.0
+apiVersion: weknora.plugin/v1
+name: { en-US: ACME OCR }
+publisher: { id: acme }
+runtime: { type: host, kind: binary, entry: bin/ocr }
+contributes:
+  parsers:
+    - id: ocr
+      name: { en-US: ACME OCR, zh-CN: ACME 识别 }
+      fileTypes: [tiff, pdf]
+`
+
+func TestPluginParser(t *testing.T) {
+	ctx := context.Background()
+	plugin := pluginsdk.New(pluginsdk.Info{ID: "acme.ocr", Version: "1.0.0"})
+	plugin.Parser("ocr", pluginsdk.ParserFunc(
+		func(_ context.Context, _ *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
+			switch string(in.Content) {
+			case "busy":
+				return nil, pluginapi.Errorf(pluginapi.CodeRateLimited, "slow down")
+			case "broken":
+				return nil, pluginapi.Errorf(pluginapi.CodeInvalidConfig, "not a scanned document")
+			}
+			return &pluginapi.ParseOutput{
+				Markdown: "# " + in.FileName + "\n\n![scan](page-1.png)",
+				Images: []pluginapi.ParsedImage{
+					{OriginalRef: "page-1.png", MimeType: "image/png", Data: []byte{1, 2}},
+				},
+				Metadata: map[string]string{"pages": "1"},
+			}, nil
+		}))
+	srv := httptest.NewServer(plugin.Handler())
+	defer srv.Close()
+
+	p, err := pkg.Open(plugintest.Zip(t, map[string]string{"plugin.yaml": ocrManifest, "bin/ocr": "x"}))
+	if err != nil {
+		t.Fatal(err)
+	}
+	a := NewParsers(NewInvoker(fakeClients{client.New(srv.URL, nil, nil)}))
+	if err := a.Activate(ctx, &reconcile.Loaded{Manifest: p.Manifest, Package: p}); err != nil {
+		t.Fatalf("Activate: %v", err)
+	}
+	defer func() { _ = a.Deactivate(ctx, "acme.ocr") }()
+
+	var listed *types.ParserEngineInfo
+	for _, e := range docparser.ListAllEngines(false, nil, nil) {
+		if e.Name == "acme.ocr/ocr" {
+			listed = &e
+		}
+	}
+	if listed == nil || !listed.Available || listed.PluginID != "acme.ocr" ||
+		listed.DisplayNames["zh-CN"] != "ACME 识别" ||
+		!slices.Equal(listed.FileTypes, []string{"tiff", "pdf"}) {
+		t.Fatalf("listed = %+v", listed)
+	}
+	if !slices.Contains(docparser.PluginFileTypes(), "tiff") {
+		t.Fatal("plugin file types must be importable")
+	}
+
+	reader, err := docparser.NewReader(ctx, "acme.ocr/ocr", "tiff", false, docparser.ReaderDeps{
+		Overrides: map[string]string{"mineru_api_key": "must-not-leak"},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	res, err := reader.Read(
+		ctx,
+		&types.ReadRequest{FileName: "scan.tiff", FileType: ".TIFF", FileContent: []byte("img")},
+	)
+	if err != nil || !strings.HasPrefix(res.MarkdownContent, "# scan.tiff") || res.Metadata["pages"] != "1" ||
+		len(
+			res.ImageRefs,
+		) != 1 || res.ImageRefs[0].OriginalRef != "page-1.png" || res.ImageRefs[0].Filename != "page-1.png" {
+		t.Fatalf("Read = %+v, %v", res, err)
+	}
+
+	if _, err := reader.Read(ctx, &types.ReadRequest{FileContent: []byte("busy")}); err == nil {
+		t.Fatal("a retryable plugin error must come back as an error, so the task retries")
+	}
+	res, err = reader.Read(ctx, &types.ReadRequest{FileContent: []byte("broken")})
+	if err != nil || res.Error != "not a scanned document" {
+		t.Fatalf("a permanent plugin error must fail the document: %+v, %v", res, err)
+	}
+
+	if err := a.Deactivate(ctx, "acme.ocr"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := docparser.NewReader(ctx, "acme.ocr/ocr", "tiff", false, docparser.ReaderDeps{}); err == nil ||
+		!strings.Contains(err.Error(), "not installed or not running") {
+		t.Fatalf("a removed plugin engine must not fall through to the docreader: %v", err)
+	}
+}
+
+func TestPluginParserCannotShadowABuiltin(t *testing.T) {
+	e := &pluginEngine{name: "simple"}
+	if err := docparser.RegisterPluginEngine(e); err == nil {
+		t.Fatal("a plugin engine must not take a builtin engine's name")
+	}
+}
diff --git a/internal/plugin/host/process.go b/internal/plugin/host/process.go
index e40eb28f3..ed2bbde95 100644
--- a/internal/plugin/host/process.go
+++ b/internal/plugin/host/process.go
@@ -167,8 +167,10 @@ func (p *process) childEnv(socket, token string) []string {
 		"HTTPS_PROXY=" + p.proxy.URL(),
 		"http_proxy=" + p.proxy.URL(),
 		"https_proxy=" + p.proxy.URL(),
-		"NO_PROXY=",
-		"no_proxy=",
+		// The Host API is on this node's loopback; the proxy is for the
+		// outside world.
+		"NO_PROXY=127.0.0.1,localhost,::1",
+		"no_proxy=127.0.0.1,localhost,::1",
 	}
 	for _, k := range []string{"PATH", "LANG", "LC_ALL", "TZ", "SYSTEMROOT", "WINDIR"} {
 		if v, ok := os.LookupEnv(k); ok {
diff --git a/internal/plugin/hostapi/hostapi_test.go b/internal/plugin/hostapi/hostapi_test.go
new file mode 100644
index 000000000..cf5838a9a
--- /dev/null
+++ b/internal/plugin/hostapi/hostapi_test.go
@@ -0,0 +1,111 @@
+package hostapi
+
+import (
+	"context"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"github.com/Tencent/WeKnora/internal/application/repository"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/pluginsdk"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+func TestTokens(t *testing.T) {
+	iss := NewIssuer([]byte("k"))
+	tok, exp, err := iss.Issue("acme.x", "1.0.0", 7, []string{"kv"})
+	require.NoError(t, err)
+	require.WithinDuration(t, time.Now().Add(TokenTTL), exp, time.Second)
+	c, err := iss.Verify(tok)
+	require.NoError(t, err)
+	require.Equal(t, uint64(7), c.TenantID)
+	require.True(t, c.Has("kv"))
+
+	_, err = NewIssuer([]byte("other")).Verify(tok)
+	require.ErrorIs(t, err, ErrInvalidToken, "another key must not verify")
+	later := NewIssuer([]byte("k"))
+	later.now = func() time.Time { return time.Now().Add(TokenTTL + time.Minute) }
+	_, err = later.Verify(tok)
+	require.ErrorIs(t, err, ErrInvalidToken, "expired tokens must not verify")
+	_, err = iss.Verify(tok[:len(tok)-2] + "xx")
+	require.ErrorIs(t, err, ErrInvalidToken)
+}
+
+// hostCall is what a plugin's call would carry, pointing at a test server.
+func hostCall(t *testing.T, url string, iss *Issuer, tenant uint64, scopes ...string) *pluginsdk.Call {
+	t.Helper()
+	tok, _, err := iss.Issue("acme.x", "1.0.0", tenant, scopes)
+	require.NoError(t, err)
+	return &pluginsdk.Call{
+		Context: pluginapi.Context{TenantID: tenant, Host: &pluginapi.HostAccess{URL: url, Token: tok}},
+	}
+}
+
+func TestKVThroughTheSDK(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
+	require.NoError(t, err)
+	require.NoError(t, db.AutoMigrate(&types.PluginKV{}))
+	iss := NewIssuer([]byte("k"))
+	h := NewHandler(iss, NewKV(repository.NewPluginKVRepository(db)))
+	r := gin.New()
+	h.Register(r)
+	srv := httptest.NewServer(r)
+	defer srv.Close()
+	ctx := context.Background()
+
+	host := hostCall(t, srv.URL, iss, 1, "kv").Host()
+	require.NotNil(t, host)
+	type cursor struct{ Page int }
+	ok, err := host.KVGet(ctx, "cursor/a", &cursor{})
+	require.NoError(t, err)
+	require.False(t, ok)
+	require.NoError(t, host.KVPut(ctx, "cursor/a", cursor{Page: 3}, 0))
+	require.NoError(t, host.KVPut(ctx, "cursor/b", cursor{Page: 4}, time.Hour))
+	var got cursor
+	ok, err = host.KVGet(ctx, "cursor/a", &got)
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.Equal(t, 3, got.Page)
+
+	page, err := host.KVList(ctx, "cursor/", "", 1)
+	require.NoError(t, err)
+	require.Len(t, page.Entries, 1)
+	require.Equal(t, "cursor/a", page.Next)
+	page, err = host.KVList(ctx, "cursor/", page.Next, 1)
+	require.NoError(t, err)
+	require.Equal(t, "cursor/b", page.Entries[0].Key)
+	require.NotNil(t, page.Entries[0].ExpiresAt)
+
+	// Another tenant sees nothing of tenant 1.
+	other := hostCall(t, srv.URL, iss, 2, "kv").Host()
+	ok, err = other.KVGet(ctx, "cursor/a", &got)
+	require.NoError(t, err)
+	require.False(t, ok)
+
+	require.NoError(t, host.KVDelete(ctx, "cursor/a"))
+	ok, _ = host.KVGet(ctx, "cursor/a", &got)
+	require.False(t, ok)
+
+	// Without the scope, or with a forged token, the call is refused.
+	noScope := hostCall(t, srv.URL, iss, 1).Host()
+	_, err = noScope.KVGet(ctx, "cursor/b", &got)
+	require.ErrorContains(t, err, `not granted the "kv" scope`)
+	forged := hostCall(t, srv.URL, NewIssuer([]byte("wrong")), 1, "kv").Host()
+	_, err = forged.KVGet(ctx, "cursor/b", &got)
+	require.ErrorContains(t, err, "invalid or expired")
+
+	// Bad input is a bad request, not a crash.
+	err = host.KVPut(ctx, strings.Repeat("k", 300), 1, 0)
+	e, ok := pluginapi.AsError(err)
+	require.True(t, ok)
+	require.Equal(t, pluginapi.CodeBadRequest, e.Code)
+}
diff --git a/internal/plugin/hostapi/http.go b/internal/plugin/hostapi/http.go
new file mode 100644
index 000000000..423df13a6
--- /dev/null
+++ b/internal/plugin/hostapi/http.go
@@ -0,0 +1,125 @@
+package hostapi
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/Tencent/WeKnora/internal/logger"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// PathPrefix is where the Host API is served. The global auth middleware
+// lets it through; Handler authenticates plugin tokens itself.
+const PathPrefix = "/api/v1/plugin-host/"
+
+const claimsKey = "pluginHostClaims"
+
+// Handler serves the Host API.
+type Handler struct {
+	issuer *Issuer
+	kv     *KV
+}
+
+// NewHandler creates the Host API handler.
+func NewHandler(issuer *Issuer, kv *KV) *Handler { return &Handler{issuer: issuer, kv: kv} }
+
+func writeErr(c *gin.Context, err error) {
+	var e *pluginapi.Error
+	if !errors.As(err, &e) {
+		logger.Errorf(c.Request.Context(), "[plugin] host api %s: %v", c.Request.URL.Path, err)
+		e = &pluginapi.Error{Code: pluginapi.CodeInternal, Message: "host api failed"}
+	}
+	c.AbortWithStatusJSON(e.Code.HTTPStatus(), pluginapi.ErrorBody{Error: *e})
+}
+
+// authenticate verifies the plugin token and the scope the route needs.
+func (h *Handler) authenticate(scope string) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		token, ok := strings.CutPrefix(c.GetHeader("Authorization"), "Bearer ")
+		if !ok || token == "" {
+			writeErr(c, pluginapi.Errorf(pluginapi.CodeUnauthorized, "missing plugin token"))
+			return
+		}
+		claims, err := h.issuer.Verify(token)
+		if err != nil {
+			writeErr(c, pluginapi.Errorf(pluginapi.CodeUnauthorized, "invalid or expired plugin token"))
+			return
+		}
+		if !claims.Has(scope) {
+			writeErr(
+				c,
+				pluginapi.Errorf(pluginapi.CodeUnauthorized, "the plugin was not granted the %q scope", scope),
+			)
+			return
+		}
+		c.Set(claimsKey, claims)
+		c.Next()
+	}
+}
+
+func claimsOf(c *gin.Context) *Claims {
+	v, _ := c.Get(claimsKey)
+	claims, _ := v.(*Claims)
+	return claims
+}
+
+// Register mounts the Host API on the engine root.
+func (h *Handler) Register(r gin.IRouter) {
+	g := r.Group(strings.TrimSuffix(PathPrefix, "/"))
+	kv := g.Group("/kv", h.authenticate(ScopeKV))
+	kv.GET("", h.kvGet)
+	kv.PUT("", h.kvPut)
+	kv.DELETE("", h.kvDelete)
+	kv.GET("/list", h.kvList)
+}
+
+func (h *Handler) kvGet(c *gin.Context) {
+	e, err := h.kv.Get(c.Request.Context(), claimsOf(c), c.Query("key"))
+	if err != nil {
+		writeErr(c, err)
+		return
+	}
+	c.JSON(http.StatusOK, e)
+}
+
+func (h *Handler) kvPut(c *gin.Context) {
+	var in pluginapi.KVPut
+	body, err := io.ReadAll(io.LimitReader(c.Request.Body, pluginapi.KVMaxValueBytes+pluginapi.KVMaxKeyBytes+1024))
+	if err == nil {
+		err = json.Unmarshal(body, &in)
+	}
+	if err != nil {
+		writeErr(c, pluginapi.Errorf(pluginapi.CodeBadRequest, "body must be {key, value, ttlSeconds}"))
+		return
+	}
+	e, err := h.kv.Put(c.Request.Context(), claimsOf(c), in)
+	if err != nil {
+		writeErr(c, err)
+		return
+	}
+	c.JSON(http.StatusOK, e)
+}
+
+func (h *Handler) kvDelete(c *gin.Context) {
+	if err := h.kv.Delete(c.Request.Context(), claimsOf(c), c.Query("key")); err != nil {
+		writeErr(c, err)
+		return
+	}
+	c.Status(http.StatusNoContent)
+}
+
+func (h *Handler) kvList(c *gin.Context) {
+	limit, _ := strconv.Atoi(c.Query("limit"))
+	out, err := h.kv.List(c.Request.Context(), claimsOf(c), c.Query("prefix"), c.Query("after"), limit)
+	if err != nil {
+		writeErr(c, err)
+		return
+	}
+	c.JSON(http.StatusOK, out)
+}
diff --git a/internal/plugin/hostapi/kv.go b/internal/plugin/hostapi/kv.go
new file mode 100644
index 000000000..c5ebed55b
--- /dev/null
+++ b/internal/plugin/hostapi/kv.go
@@ -0,0 +1,162 @@
+package hostapi
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"time"
+	"unicode"
+	"unicode/utf8"
+
+	"github.com/Tencent/WeKnora/internal/logger"
+	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/Tencent/WeKnora/internal/types/interfaces"
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// ScopeKV grants a plugin its own key-value store in the calling tenant.
+const ScopeKV = "kv"
+
+// KV is the plugins' key-value store. Every operation is confined to the
+// plugin and tenant of the token, so a plugin never sees another plugin's
+// or another workspace's data.
+type KV struct {
+	repo interfaces.PluginKVRepository
+	now  func() time.Time
+}
+
+// NewKV creates the store.
+func NewKV(repo interfaces.PluginKVRepository) *KV { return &KV{repo: repo, now: time.Now} }
+
+func checkKey(key string) error {
+	switch {
+	case key == "":
+		return pluginapi.Errorf(pluginapi.CodeBadRequest, "key is required")
+	case len(key) > pluginapi.KVMaxKeyBytes:
+		return pluginapi.Errorf(pluginapi.CodeBadRequest, "key is over %d bytes", pluginapi.KVMaxKeyBytes)
+	case !utf8.ValidString(key) || strings.ContainsFunc(key, unicode.IsControl):
+		return pluginapi.Errorf(pluginapi.CodeBadRequest, "key must be printable UTF-8")
+	}
+	return nil
+}
+
+func toEntry(e *types.PluginKV) pluginapi.KVEntry {
+	return pluginapi.KVEntry{
+		Key:       e.Key,
+		Value:     json.RawMessage(e.Value),
+		ExpiresAt: e.ExpiresAt,
+		UpdatedAt: e.UpdatedAt,
+	}
+}
+
+// Get reads a key.
+func (s *KV) Get(ctx context.Context, c *Claims, key string) (*pluginapi.KVEntry, error) {
+	if err := checkKey(key); err != nil {
+		return nil, err
+	}
+	e, err := s.repo.Get(ctx, c.PluginID, c.TenantID, key)
+	if err != nil {
+		return nil, err
+	}
+	if e == nil {
+		return nil, pluginapi.Errorf(pluginapi.CodeNotFound, "no key %q", key)
+	}
+	out := toEntry(e)
+	return &out, nil
+}
+
+// Put writes a key, refusing new keys past the per-tenant quota.
+func (s *KV) Put(ctx context.Context, c *Claims, in pluginapi.KVPut) (*pluginapi.KVEntry, error) {
+	if err := checkKey(in.Key); err != nil {
+		return nil, err
+	}
+	if len(in.Value) == 0 || !json.Valid(in.Value) {
+		return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "value must be JSON")
+	}
+	if len(in.Value) > pluginapi.KVMaxValueBytes {
+		return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "value is over %d bytes", pluginapi.KVMaxValueBytes)
+	}
+	if in.TTLSeconds < 0 {
+		return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "ttlSeconds must not be negative")
+	}
+	existing, err := s.repo.Get(ctx, c.PluginID, c.TenantID, in.Key)
+	if err != nil {
+		return nil, err
+	}
+	if existing == nil {
+		n, err := s.repo.Count(ctx, c.PluginID, c.TenantID)
+		if err != nil {
+			return nil, err
+		}
+		if n >= pluginapi.KVMaxKeys {
+			return nil, pluginapi.Errorf(
+				pluginapi.CodeRateLimited,
+				"the store holds its maximum of %d keys",
+				pluginapi.KVMaxKeys,
+			)
+		}
+	}
+	now := s.now()
+	e := &types.PluginKV{
+		PluginID: c.PluginID, TenantID: c.TenantID, Key: in.Key, Value: types.JSON(in.Value), UpdatedAt: now,
+	}
+	if in.TTLSeconds > 0 {
+		exp := now.Add(time.Duration(in.TTLSeconds) * time.Second)
+		e.ExpiresAt = &exp
+	}
+	if err := s.repo.Put(ctx, e); err != nil {
+		return nil, err
+	}
+	out := toEntry(e)
+	return &out, nil
+}
+
+// Delete removes a key; deleting a missing key is not an error.
+func (s *KV) Delete(ctx context.Context, c *Claims, key string) error {
+	if err := checkKey(key); err != nil {
+		return err
+	}
+	_, err := s.repo.Delete(ctx, c.PluginID, c.TenantID, key)
+	return err
+}
+
+// List pages through keys with a prefix.
+func (s *KV) List(ctx context.Context, c *Claims, prefix, after string, limit int) (*pluginapi.KVList, error) {
+	if limit <= 0 {
+		limit = 100
+	}
+	limit = min(limit, pluginapi.KVMaxListLimit)
+	rows, err := s.repo.List(ctx, c.PluginID, c.TenantID, prefix, after, limit+1)
+	if err != nil {
+		return nil, err
+	}
+	out := &pluginapi.KVList{Entries: []pluginapi.KVEntry{}}
+	for i := range rows {
+		if i == limit {
+			out.Next = rows[i-1].Key
+			break
+		}
+		out.Entries = append(out.Entries, toEntry(&rows[i]))
+	}
+	return out, nil
+}
+
+// StartSweeper deletes expired entries every interval until ctx ends.
+func (s *KV) StartSweeper(ctx context.Context, interval time.Duration) {
+	go func() {
+		t := time.NewTicker(interval)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				if n, err := s.repo.DeleteExpired(ctx, s.now()); err != nil {
+					logger.Warnf(ctx, "[plugin] sweep expired kv: %v", err)
+				} else if n > 0 {
+					logger.Infof(ctx, "[plugin] swept %d expired kv entries", n)
+				}
+			}
+		}
+	}()
+}
diff --git a/internal/plugin/hostapi/token.go b/internal/plugin/hostapi/token.go
new file mode 100644
index 000000000..4449166e1
--- /dev/null
+++ b/internal/plugin/hostapi/token.go
@@ -0,0 +1,110 @@
+// Package hostapi is the Host API: how a code plugin calls back into
+// WeKnora. Every call to a plugin that was granted Host API scopes carries a
+// short-lived token bound to the plugin, the tenant and those scopes; the
+// plugin presents it to /api/v1/plugin-host/*.
+package hostapi
+
+import (
+	"crypto/hmac"
+	"crypto/rand"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"os"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/golang-jwt/jwt/v5"
+
+	"github.com/Tencent/WeKnora/internal/utils"
+)
+
+// Audience of Host API tokens; they are refused anywhere else.
+const Audience = "weknora-host-api"
+
+// TokenTTL is how long a token lives: one call and its follow-ups.
+const TokenTTL = 5 * time.Minute
+
+// Claims are what a Host API token asserts.
+type Claims struct {
+	PluginID string   `json:"plugin"`
+	Version  string   `json:"ver"`
+	TenantID uint64   `json:"tenant"`
+	Scopes   []string `json:"scopes"`
+	jwt.RegisteredClaims
+}
+
+// Has reports whether the token grants a scope.
+func (c *Claims) Has(scope string) bool { return slices.Contains(c.Scopes, scope) }
+
+// Issuer signs and verifies Host API tokens.
+type Issuer struct {
+	key []byte
+	now func() time.Time
+}
+
+// NewIssuer signs with key. Nodes must share it for a token issued on one
+// node to verify on another.
+func NewIssuer(key []byte) *Issuer { return &Issuer{key: key, now: time.Now} }
+
+// NewIssuerFromEnv derives the signing key from the cluster-wide secrets:
+// SYSTEM_AES_KEY, else JWT_SECRET. Without either it uses a random key, which
+// only works while plugins call back the node that called them (the
+// embedded host always does).
+func NewIssuerFromEnv() *Issuer {
+	var secret []byte
+	switch {
+	case utils.GetAESKey() != nil:
+		secret = utils.GetAESKey()
+	case strings.TrimSpace(os.Getenv("JWT_SECRET")) != "":
+		secret = []byte(strings.TrimSpace(os.Getenv("JWT_SECRET")))
+	default:
+		secret = make([]byte, 32)
+		_, _ = rand.Read(secret)
+	}
+	mac := hmac.New(sha256.New, secret)
+	_, _ = mac.Write([]byte("weknora plugin host api tokens"))
+	return NewIssuer(mac.Sum(nil))
+}
+
+// Issue signs a token for one call.
+func (i *Issuer) Issue(pluginID, version string, tenantID uint64, scopes []string) (string, time.Time, error) {
+	now := i.now()
+	exp := now.Add(TokenTTL)
+	claims := Claims{
+		PluginID: pluginID, Version: version, TenantID: tenantID, Scopes: scopes,
+		RegisteredClaims: jwt.RegisteredClaims{
+			Subject:   "plugin:" + pluginID,
+			Audience:  jwt.ClaimStrings{Audience},
+			IssuedAt:  jwt.NewNumericDate(now),
+			NotBefore: jwt.NewNumericDate(now.Add(-30 * time.Second)),
+			ExpiresAt: jwt.NewNumericDate(exp),
+		},
+	}
+	tok, err := jwt.NewWithClaims(jwt.SigningMethodHS256, claims).SignedString(i.key)
+	return tok, exp, err
+}
+
+// ErrInvalidToken is any token that does not verify.
+var ErrInvalidToken = errors.New("invalid plugin token")
+
+// Verify checks a token's signature, audience and lifetime.
+func (i *Issuer) Verify(token string) (*Claims, error) {
+	var c Claims
+	parsed, err := jwt.ParseWithClaims(token, &c, func(*jwt.Token) (any, error) {
+		return i.key, nil
+	},
+		jwt.WithValidMethods([]string{jwt.SigningMethodHS256.Alg()}),
+		jwt.WithAudience(Audience),
+		jwt.WithExpirationRequired(),
+		jwt.WithTimeFunc(i.now),
+	)
+	if err != nil || !parsed.Valid {
+		return nil, fmt.Errorf("%w: %v", ErrInvalidToken, err)
+	}
+	if c.PluginID == "" || c.TenantID == 0 || c.Subject != "plugin:"+c.PluginID {
+		return nil, fmt.Errorf("%w: incomplete claims", ErrInvalidToken)
+	}
+	return &c, nil
+}
diff --git a/internal/plugin/manifest/manifest.go b/internal/plugin/manifest/manifest.go
index cc5f3fec3..14aca8569 100644
--- a/internal/plugin/manifest/manifest.go
+++ b/internal/plugin/manifest/manifest.go
@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"regexp"
+	"slices"
 	"strings"
 
 	"golang.org/x/mod/semver"
@@ -150,6 +151,8 @@ type Contribution struct {
 	Path string `json:"path,omitempty"           yaml:"path"`
 	// MCP describes a remote MCP server (mcpServers).
 	MCP *MCPServer `json:"mcp,omitempty"            yaml:"mcp"`
+	// FileTypes are the lower-case extensions a parser handles ("pdf").
+	FileTypes []string `json:"fileTypes,omitempty"      yaml:"fileTypes"`
 	// Extra carries point-specific metadata the generic fields do not cover.
 	Extra map[string]any `json:"extra,omitempty"          yaml:"extra"`
 }
@@ -287,7 +290,19 @@ var egressPattern = regexp.MustCompile(`^(\*\.)?([a-z0-9]([a-z0-9-]*[a-z0-9])?)(
 // follows links anywhere). Private addresses stay unreachable either way.
 const EgressAnyHost = "*"
 
+// HostAPIScopes are the Host API scopes a plugin may ask for in
+// permissions.hostApi.
+var HostAPIScopes = []string{
+	// kv: the plugin's own key-value store in each tenant.
+	"kv",
+}
+
 func (m *Manifest) validatePermissions(add func(string, ...any)) {
+	for _, s := range m.Permissions.HostAPI {
+		if !slices.Contains(HostAPIScopes, s) {
+			add("permissions.hostApi %q is not a Host API scope (known: %s)", s, strings.Join(HostAPIScopes, ", "))
+		}
+	}
 	for _, e := range m.Permissions.Egress {
 		if e != EgressAnyHost && !egressPattern.MatchString(strings.ToLower(e)) {
 			add(
@@ -345,6 +360,9 @@ func (m *Manifest) validateContributions(add func(string, ...any)) {
 				add("%s.aliases may only be declared by builtin plugins", where)
 			}
 			validateDeclarative(point, c, m.Builtin, where, add)
+			if point == PointParsers && !m.Builtin {
+				validateFileTypes(c.FileTypes, where, add)
+			}
 			for _, alias := range c.Aliases {
 				// Aliases share the ID alphabet, which has no '/', so an
 				// alias can never shadow a qualified ID.
@@ -358,6 +376,20 @@ func (m *Manifest) validateContributions(add func(string, ...any)) {
 
 // validateDeclarative checks the fields a declarative contribution needs.
 // Builtins describe their implementation in code instead.
+// fileTypePattern is a lower-case file extension without the dot.
+var fileTypePattern = regexp.MustCompile(`^[a-z0-9]{1,16}$`)
+
+func validateFileTypes(types []string, where string, add func(string, ...any)) {
+	if len(types) == 0 {
+		add("%s.fileTypes is required: the extensions the parser handles, such as [pdf]", where)
+	}
+	for _, t := range types {
+		if !fileTypePattern.MatchString(t) {
+			add("%s.fileTypes %q must be a lower-case extension without the dot", where, t)
+		}
+	}
+}
+
 func validateDeclarative(point Point, c Contribution, builtin bool, where string, add func(string, ...any)) {
 	if builtin {
 		return
diff --git a/internal/plugin/manifest/point.go b/internal/plugin/manifest/point.go
index f3c90ccb1..bce0b6204 100644
--- a/internal/plugin/manifest/point.go
+++ b/internal/plugin/manifest/point.go
@@ -40,7 +40,7 @@ var points = []PointInfo{
 	{Point: PointIMChannels},
 	{Point: PointWebSearch, ThirdParty: true},
 	{Point: PointTools},
-	{Point: PointParsers},
+	{Point: PointParsers, ThirdParty: true},
 	{Point: PointSkills, ThirdParty: true, Declarative: true},
 	{Point: PointMCPServers, ThirdParty: true, Declarative: true},
 }
diff --git a/internal/router/router.go b/internal/router/router.go
index 99e022e3b..b7a241d5c 100644
--- a/internal/router/router.go
+++ b/internal/router/router.go
@@ -20,6 +20,7 @@ import (
 	"github.com/Tencent/WeKnora/internal/logger"
 	"github.com/Tencent/WeKnora/internal/mcpserver"
 	"github.com/Tencent/WeKnora/internal/middleware"
+	"github.com/Tencent/WeKnora/internal/plugin/hostapi"
 	"github.com/Tencent/WeKnora/internal/tracing/langfuse"
 	"github.com/Tencent/WeKnora/internal/types/interfaces"
 
@@ -92,6 +93,7 @@ type RouterParams struct {
 	DataSourceCredentialsHandler *handler.DataSourceCredentialsHandler
 	PluginHandler                *handler.PluginHandler
 	PluginAdminHandler           *handler.PluginAdminHandler
+	PluginHostAPI                *hostapi.Handler
 	WeKnoraCloudHandler          *handler.WeKnoraCloudHandler
 	WikiPageHandler              *handler.WikiPageHandler
 	MemoryHandler                *handler.MemoryHandler
@@ -206,6 +208,12 @@ func NewRouter(params RouterParams) *gin.Engine {
 	r.POST("/api/v1/local-browser/extension/authorize", params.SessionHandler.BrowserSkillAuthorize)
 	r.POST("/api/v1/local-browser/internal", params.SessionHandler.BrowserSkillInternal)
 
+	// Host API: plugins call back with their own short-lived tokens, which
+	// the handler verifies; user and API key auth do not apply.
+	if params.PluginHostAPI != nil {
+		params.PluginHostAPI.Register(r)
+	}
+
 	// 认证中间件
 	r.Use(middleware.Auth(params.TenantService, params.UserService, params.TenantMemberService, params.TenantAPIKeyService, params.Config))
 
diff --git a/internal/types/docparser.go b/internal/types/docparser.go
index bd780c1cc..f9d2de673 100644
--- a/internal/types/docparser.go
+++ b/internal/types/docparser.go
@@ -51,6 +51,10 @@ type ParserEngineInfo struct {
 	FileTypes         []string
 	Available         bool
 	UnavailableReason string
+	// PluginID and DisplayNames describe an engine an installed plugin
+	// provides; DisplayNames is keyed by locale.
+	PluginID     string            `json:",omitempty"`
+	DisplayNames map[string]string `json:",omitempty"`
 }
 
 // --- Internal types used by chunking pipeline ---
diff --git a/internal/types/interfaces/plugin.go b/internal/types/interfaces/plugin.go
index 6028f71fc..1fce016e5 100644
--- a/internal/types/interfaces/plugin.go
+++ b/internal/types/interfaces/plugin.go
@@ -2,6 +2,7 @@ package interfaces
 
 import (
 	"context"
+	"time"
 
 	"github.com/Tencent/WeKnora/internal/plugin/manifest"
 	"github.com/Tencent/WeKnora/internal/types"
@@ -41,3 +42,26 @@ type PluginRepository interface {
 	GetVersion(ctx context.Context, pluginID, version string) (*types.PluginVersion, error)
 	SaveVersion(ctx context.Context, v *types.PluginVersion) error
 }
+
+// PluginKVRepository stores plugins' Host API key-value data. Expired
+// entries read as missing.
+type PluginKVRepository interface {
+	// Get returns (nil, nil) for a missing or expired key.
+	Get(ctx context.Context, pluginID string, tenantID uint64, key string) (*types.PluginKV, error)
+	Put(ctx context.Context, e *types.PluginKV) error
+	// Delete reports whether the key existed.
+	Delete(ctx context.Context, pluginID string, tenantID uint64, key string) (bool, error)
+	// List returns live keys starting with prefix, after the given key, in
+	// key order.
+	List(
+		ctx context.Context,
+		pluginID string,
+		tenantID uint64,
+		prefix, after string,
+		limit int,
+	) ([]types.PluginKV, error)
+	// Count counts live keys of a plugin in a tenant.
+	Count(ctx context.Context, pluginID string, tenantID uint64) (int64, error)
+	// DeleteExpired removes entries that expired before now.
+	DeleteExpired(ctx context.Context, now time.Time) (int64, error)
+}
diff --git a/internal/types/plugin.go b/internal/types/plugin.go
index 5b69792be..c80ff2bb8 100644
--- a/internal/types/plugin.go
+++ b/internal/types/plugin.go
@@ -48,3 +48,17 @@ type PluginVersion struct {
 
 // TableName pins the table so GORM's pluralizer cannot drift.
 func (PluginVersion) TableName() string { return "plugin_versions" }
+
+// PluginKV is one entry of a plugin's key-value store (Host API kv),
+// partitioned by plugin and tenant.
+type PluginKV struct {
+	PluginID  string     `json:"-"         gorm:"type:varchar(128);primaryKey"`
+	TenantID  uint64     `json:"-"         gorm:"primaryKey"`
+	Key       string     `json:"key"       gorm:"type:varchar(256);primaryKey"`
+	Value     JSON       `json:"value"     gorm:"type:json"`
+	ExpiresAt *time.Time `json:"expiresAt,omitempty"`
+	UpdatedAt time.Time  `json:"updatedAt"`
+}
+
+// TableName pins the table so GORM's pluralizer cannot drift.
+func (PluginKV) TableName() string { return "plugin_kv" }
diff --git a/migrations/sqlite/000036_plugin_kv.down.sql b/migrations/sqlite/000036_plugin_kv.down.sql
new file mode 100644
index 000000000..e9a7aae49
--- /dev/null
+++ b/migrations/sqlite/000036_plugin_kv.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS plugin_kv;
diff --git a/migrations/sqlite/000036_plugin_kv.up.sql b/migrations/sqlite/000036_plugin_kv.up.sql
new file mode 100644
index 000000000..367cad7fc
--- /dev/null
+++ b/migrations/sqlite/000036_plugin_kv.up.sql
@@ -0,0 +1,12 @@
+-- Host API key-value store of plugins (versioned 000117).
+CREATE TABLE IF NOT EXISTS plugin_kv (
+    plugin_id  VARCHAR(128) NOT NULL,
+    tenant_id  BIGINT       NOT NULL,
+    key        VARCHAR(256) NOT NULL,
+    value      TEXT         NOT NULL,
+    expires_at DATETIME,
+    updated_at DATETIME     NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    PRIMARY KEY (plugin_id, tenant_id, key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_plugin_kv_expires_at ON plugin_kv (expires_at);
diff --git a/migrations/versioned/000117_plugin_kv.down.sql b/migrations/versioned/000117_plugin_kv.down.sql
new file mode 100644
index 000000000..e9a7aae49
--- /dev/null
+++ b/migrations/versioned/000117_plugin_kv.down.sql
@@ -0,0 +1 @@
+DROP TABLE IF EXISTS plugin_kv;
diff --git a/migrations/versioned/000117_plugin_kv.up.sql b/migrations/versioned/000117_plugin_kv.up.sql
new file mode 100644
index 000000000..8b4d2aee1
--- /dev/null
+++ b/migrations/versioned/000117_plugin_kv.up.sql
@@ -0,0 +1,18 @@
+-- Migration 000117: the key-value store plugins use through the Host API.
+-- Plugins keep no database of their own; their state lives here, partitioned
+-- by plugin and tenant so a plugin cannot read another workspace's data.
+DO $$ BEGIN RAISE NOTICE '[Migration 000117] Creating plugin_kv'; END $$;
+
+CREATE TABLE IF NOT EXISTS plugin_kv (
+    plugin_id  VARCHAR(128) NOT NULL,
+    tenant_id  BIGINT       NOT NULL,
+    key        VARCHAR(256) NOT NULL,
+    value      JSONB        NOT NULL,
+    expires_at TIMESTAMPTZ,
+    updated_at TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
+    PRIMARY KEY (plugin_id, tenant_id, key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_plugin_kv_expires_at ON plugin_kv (expires_at) WHERE expires_at IS NOT NULL;
+
+COMMENT ON TABLE plugin_kv IS 'Host API key-value store of plugins, per plugin and tenant.';
diff --git a/pluginsdk/README.md b/pluginsdk/README.md
index 6cd65e16b..8ce07130b 100644
--- a/pluginsdk/README.md
+++ b/pluginsdk/README.md
@@ -1,7 +1,7 @@
 # WeKnora plugin SDK (Go)
 
-Build code plugins for WeKnora: web search providers and data source
-connectors that run as their own process. The module has no dependencies.
+Build code plugins for WeKnora: web search providers, data source
+connectors and document parsers that run as their own process. The module has no dependencies.
 
 | Package | What it is |
 | --- | --- |
@@ -69,6 +69,31 @@ decrypted, in three separate scopes:
 
 Keep plugins stateless and never cache a tenant's credentials across calls.
 
+## Parsers
+
+`p.Parser(id, ...)` turns a document (`in.Content`, base64 on the wire) into
+Markdown. WeKnora chunks the text and stores the images the Markdown
+references: return them in `Images` with an `OriginalRef` matching the
+`![](...)` target. Declare the extensions it handles in `plugin.yaml`
+(`fileTypes: [srt, vtt]`). Return a retryable error (`unavailable`,
+`rate_limited`) to have the document retried later; any other error fails
+it for good.
+
+## Calling back into WeKnora (Host API)
+
+A plugin that declares `permissions.hostApi` gets a short-lived token with
+each call; `call.Host()` returns a client for it, or nil without a grant.
+Scopes:
+
+- `kv`: a key-value store of the plugin, separate per workspace, so the
+  plugin needs no database.
+  - `KVGet`, `KVPut` (optional TTL), `KVDelete` and `KVList`.
+  - Keys up to 256 bytes, JSON values up to 64 KB, 10,000 keys per
+    workspace.
+
+The token is valid for a few minutes: use `call.Host()` within the call and
+don't keep it.
+
 ## Packaging
 
 A package is a zip (`.wkp`) with `plugin.yaml` at its root:
@@ -103,7 +128,8 @@ A few rules the manifest and host enforce:
   (`acme.notes/notes`); for connectors and web search that must stay within
   50 characters.
 
-See `examples/plugins/rss` for a complete plugin and its `package.sh`.
+See `examples/plugins/rss` (a connector) and `examples/plugins/subtitles` (a
+parser using the Host API) for complete plugins with their `package.sh`.
 
 ## Testing
 
diff --git a/pluginsdk/conformance/conformance.go b/pluginsdk/conformance/conformance.go
index a24d87296..5bececec3 100644
--- a/pluginsdk/conformance/conformance.go
+++ b/pluginsdk/conformance/conformance.go
@@ -140,6 +140,15 @@ func Run(ctx context.Context, t Target) Report {
 			return protocolAnswer(err)
 		})
 	}
+	for _, id := range m.Contributes["parsers"] {
+		check("parsers/"+id+" parse answers", func(ctx context.Context) error {
+			var out pluginapi.ParseOutput
+			err := t.Client.Call(ctx, pluginapi.ParsePath(id), envelope(), pluginapi.ParseInput{
+				FileName: "conformance.txt", FileType: "txt", Content: []byte("WeKnora conformance check\n"),
+			}, &out)
+			return protocolAnswer(err)
+		})
+	}
 	for _, id := range m.Contributes["connectors"] {
 		check("connectors/"+id+" validate answers", func(ctx context.Context) error {
 			return protocolAnswer(t.Client.Call(ctx, pluginapi.ConnectorValidatePath(id), envelope(), nil, nil))
@@ -227,7 +236,13 @@ func wantErrorBody(resp *http.Response, code pluginapi.ErrorCode) error {
 		return fmt.Errorf("HTTP %d without a protocol error body: %s", resp.StatusCode, b)
 	}
 	if resp.StatusCode != code.HTTPStatus() || body.Error.Code != code {
-		return fmt.Errorf("want HTTP %d %s, got HTTP %d %s", code.HTTPStatus(), code, resp.StatusCode, body.Error.Code)
+		return fmt.Errorf(
+			"want HTTP %d %s, got HTTP %d %s",
+			code.HTTPStatus(),
+			code,
+			resp.StatusCode,
+			body.Error.Code,
+		)
 	}
 	return nil
 }
diff --git a/pluginsdk/conformance/conformance_test.go b/pluginsdk/conformance/conformance_test.go
index d4ca866ab..caa97695b 100644
--- a/pluginsdk/conformance/conformance_test.go
+++ b/pluginsdk/conformance/conformance_test.go
@@ -65,12 +65,25 @@ func TestSDKPluginConforms(t *testing.T) {
 			},
 		),
 	)
+	p.Parser(
+		"text",
+		pluginsdk.ParserFunc(
+			func(_ context.Context, _ *pluginsdk.Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
+				return &pluginapi.ParseOutput{Markdown: string(in.Content)}, nil
+			},
+		),
+	)
 	rep := conformance.Run(context.Background(), target(t, p.Handler()))
+	names := map[string]bool{}
 	for _, r := range rep.Results {
+		names[r.Name] = true
 		if !r.Passed {
 			t.Errorf("%s: %s", r.Name, r.Detail)
 		}
 	}
+	if !names["parsers/text parse answers"] {
+		t.Error("parsers must be checked")
+	}
 	if !rep.Passed() || rep.Plugin != "acme.strict@1.0.0" || len(rep.Results) < 10 {
 		t.Fatalf("report = %+v", rep)
 	}
diff --git a/pluginsdk/host.go b/pluginsdk/host.go
new file mode 100644
index 000000000..d80d57154
--- /dev/null
+++ b/pluginsdk/host.go
@@ -0,0 +1,114 @@
+package pluginsdk
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// Host calls back into WeKnora on behalf of one call: the tenant and scopes
+// are those the call's token carries. It is only valid while the token lives
+// (a few minutes), so do not keep it beyond the call.
+type Host struct {
+	base  string
+	token string
+	http  *http.Client
+}
+
+// Host returns the Host API client of this call, or nil when the plugin was
+// granted no Host API scopes.
+func (c *Call) Host() *Host {
+	if c.Context.Host == nil || c.Context.Host.URL == "" || c.Context.Host.Token == "" {
+		return nil
+	}
+	return &Host{
+		base: strings.TrimSuffix(c.Context.Host.URL, "/"), token: c.Context.Host.Token,
+		http: &http.Client{Timeout: 30 * time.Second},
+	}
+}
+
+func (h *Host) do(ctx context.Context, method, path string, query url.Values, body, out any) error {
+	var rd io.Reader
+	if body != nil {
+		b, err := json.Marshal(body)
+		if err != nil {
+			return err
+		}
+		rd = bytes.NewReader(b)
+	}
+	u := h.base + path
+	if len(query) > 0 {
+		u += "?" + query.Encode()
+	}
+	req, err := http.NewRequestWithContext(ctx, method, u, rd)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Authorization", "Bearer "+h.token)
+	if body != nil {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	resp, err := h.http.Do(req)
+	if err != nil {
+		return pluginapi.Errorf(pluginapi.CodeUnavailable, "host api: %v", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode/100 != 2 {
+		var eb pluginapi.ErrorBody
+		b, _ := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+		if json.Unmarshal(b, &eb) == nil && eb.Error.Code != "" {
+			return &eb.Error
+		}
+		return pluginapi.Errorf(pluginapi.CodeInternal, "host api answered HTTP %d", resp.StatusCode)
+	}
+	if out == nil || resp.StatusCode == http.StatusNoContent {
+		return nil
+	}
+	return json.NewDecoder(resp.Body).Decode(out)
+}
+
+// KVGet reads key into v. ok is false when the key does not exist.
+func (h *Host) KVGet(ctx context.Context, key string, v any) (ok bool, err error) {
+	var e pluginapi.KVEntry
+	err = h.do(ctx, http.MethodGet, pluginapi.HostKVPath, url.Values{"key": {key}}, nil, &e)
+	if pe, isPE := pluginapi.AsError(err); isPE && pe.Code == pluginapi.CodeNotFound {
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+	return true, json.Unmarshal(e.Value, v)
+}
+
+// KVPut stores v (as JSON) under key; ttl 0 keeps it until deleted.
+func (h *Host) KVPut(ctx context.Context, key string, v any, ttl time.Duration) error {
+	raw, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+	in := pluginapi.KVPut{Key: key, Value: raw, TTLSeconds: int(ttl / time.Second)}
+	return h.do(ctx, http.MethodPut, pluginapi.HostKVPath, nil, in, nil)
+}
+
+// KVDelete removes key; a missing key is not an error.
+func (h *Host) KVDelete(ctx context.Context, key string) error {
+	return h.do(ctx, http.MethodDelete, pluginapi.HostKVPath, url.Values{"key": {key}}, nil, nil)
+}
+
+// KVList returns a page of keys with prefix after the given key.
+func (h *Host) KVList(ctx context.Context, prefix, after string, limit int) (*pluginapi.KVList, error) {
+	q := url.Values{"prefix": {prefix}, "after": {after}}
+	if limit > 0 {
+		q.Set("limit", strconv.Itoa(limit))
+	}
+	var out pluginapi.KVList
+	return &out, h.do(ctx, http.MethodGet, pluginapi.HostKVListPath, q, nil, &out)
+}
diff --git a/pluginsdk/openapi_test.go b/pluginsdk/openapi_test.go
index 4d8a4346d..014d99df5 100644
--- a/pluginsdk/openapi_test.go
+++ b/pluginsdk/openapi_test.go
@@ -25,7 +25,7 @@ func TestOpenAPIMatchesRoutes(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	for _, f := range []string{"websearch.go", "connector.go"} {
+	for _, f := range []string{"websearch.go", "connector.go", "parser.go"} {
 		b, err := os.ReadFile(f)
 		if err != nil {
 			t.Fatal(err)
diff --git a/pluginsdk/parser.go b/pluginsdk/parser.go
new file mode 100644
index 000000000..563d2cbbd
--- /dev/null
+++ b/pluginsdk/parser.go
@@ -0,0 +1,55 @@
+package pluginsdk
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/Tencent/WeKnora/pluginsdk/pluginapi"
+)
+
+// Parser turns a document into Markdown. Return a non-retryable error (such
+// as pluginapi.CodeInvalidConfig or CodeNotFound) for a document that will
+// never parse; retryable ones (CodeUnavailable, CodeRateLimited) make
+// WeKnora try again later.
+type Parser interface {
+	Parse(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error)
+}
+
+// ParserFunc adapts a function to Parser.
+type ParserFunc func(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error)
+
+// Parse implements Parser.
+func (f ParserFunc) Parse(ctx context.Context, call *Call, in pluginapi.ParseInput) (*pluginapi.ParseOutput, error) {
+	return f(ctx, call, in)
+}
+
+// Parser registers parser id (contributes.parsers[].id).
+func (p *Plugin) Parser(id string, parser Parser) { p.parsers[id] = parser }
+
+func (p *Plugin) routeParsers(mux *http.ServeMux) {
+	mux.HandleFunc("POST /v1/parsers/{id}/parse", func(w http.ResponseWriter, r *http.Request) {
+		parser, ok := p.parsers[r.PathValue("id")]
+		if !ok {
+			writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no parser %q", r.PathValue("id")))
+			return
+		}
+		p.unary(func(ctx context.Context, call *Call, raw json.RawMessage) (any, error) {
+			in, err := decodeInput[pluginapi.ParseInput](raw)
+			if err != nil {
+				return nil, err
+			}
+			if len(in.Content) == 0 && in.URL == "" {
+				return nil, pluginapi.Errorf(pluginapi.CodeBadRequest, "content or url is required")
+			}
+			out, err := parser.Parse(ctx, call, in)
+			if err != nil {
+				return nil, err
+			}
+			if out == nil {
+				out = &pluginapi.ParseOutput{}
+			}
+			return out, nil
+		})(w, r)
+	})
+}
diff --git a/pluginsdk/plugin.go b/pluginsdk/plugin.go
index 61c044386..486986183 100644
--- a/pluginsdk/plugin.go
+++ b/pluginsdk/plugin.go
@@ -62,6 +62,7 @@ type Plugin struct {
 	info       Info
 	webSearch  map[string]WebSearcher
 	connectors map[string]Connector
+	parsers    map[string]Parser
 	validate   ConfigValidator
 	logger     *slog.Logger
 	// ShutdownTimeout bounds how long Serve waits for calls in flight after
@@ -75,6 +76,7 @@ func New(info Info) *Plugin {
 		info:            info,
 		webSearch:       map[string]WebSearcher{},
 		connectors:      map[string]Connector{},
+		parsers:         map[string]Parser{},
 		logger:          slog.New(slog.NewTextHandler(os.Stderr, nil)),
 		ShutdownTimeout: 60 * time.Second,
 	}
@@ -100,6 +102,7 @@ func (p *Plugin) Manifest() pluginapi.Manifest {
 	}
 	add("webSearch", keys(p.webSearch))
 	add("connectors", keys(p.connectors))
+	add("parsers", keys(p.parsers))
 	return m
 }
 
@@ -134,6 +137,7 @@ func (p *Plugin) Handler() http.Handler {
 	)
 	p.routeWebSearch(mux)
 	p.routeConnectors(mux)
+	p.routeParsers(mux)
 	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
 		writeError(w, pluginapi.Errorf(pluginapi.CodeNotFound, "no endpoint %s %s", r.Method, r.URL.Path))
 	})
diff --git a/pluginsdk/pluginapi/hostapi.go b/pluginsdk/pluginapi/hostapi.go
new file mode 100644
index 000000000..7fc7e827d
--- /dev/null
+++ b/pluginsdk/pluginapi/hostapi.go
@@ -0,0 +1,48 @@
+package pluginapi
+
+import (
+	"encoding/json"
+	"time"
+)
+
+// Host API: how a plugin calls back into WeKnora. The URL and a short-lived
+// bearer token arrive in Context.Host of each call; the token is bound to the
+// tenant of that call and the scopes the plugin was granted (manifest
+// permissions.hostApi). Errors use ErrorBody.
+
+// HostKVPath is the key-value endpoint (scope "kv"): GET and DELETE take
+// ?key=, PUT takes a KVPut body.
+const HostKVPath = "/api/v1/plugin-host/kv"
+
+// HostKVListPath lists keys: ?prefix=&after=&limit=.
+const HostKVListPath = "/api/v1/plugin-host/kv/list"
+
+// KVEntry is one stored value.
+type KVEntry struct {
+	Key       string          `json:"key"`
+	Value     json.RawMessage `json:"value"`
+	ExpiresAt *time.Time      `json:"expiresAt,omitempty"`
+	UpdatedAt time.Time       `json:"updatedAt"`
+}
+
+// KVPut stores a value; TTLSeconds 0 keeps it until deleted.
+type KVPut struct {
+	Key        string          `json:"key"`
+	Value      json.RawMessage `json:"value"`
+	TTLSeconds int             `json:"ttlSeconds,omitempty"`
+}
+
+// KVList is a page of keys in key order; pass Next as after for the next
+// page (empty when there is none).
+type KVList struct {
+	Entries []KVEntry `json:"entries"`
+	Next    string    `json:"next,omitempty"`
+}
+
+// Limits of the key-value store.
+const (
+	KVMaxKeyBytes   = 256
+	KVMaxValueBytes = 64 << 10
+	KVMaxKeys       = 10000 // per plugin and tenant
+	KVMaxListLimit  = 1000
+)
diff --git a/pluginsdk/pluginapi/openapi.yaml b/pluginsdk/pluginapi/openapi.yaml
index 711664c3b..30e07a57b 100644
--- a/pluginsdk/pluginapi/openapi.yaml
+++ b/pluginsdk/pluginapi/openapi.yaml
@@ -162,6 +162,33 @@ paths:
             application/x-ndjson:
               schema: { $ref: "#/components/schemas/Event" }
         default: { $ref: "#/components/responses/Error" }
+  /v1/parsers/{id}/parse:
+    post:
+      summary: Parse a document into Markdown (DocReader.Read)
+      description: |
+        The file arrives as base64 content (or a URL). Answer Markdown and
+        the images it references; WeKnora stores the images and chunks the
+        text. A non-retryable error fails the document for good; a
+        retryable one (unavailable, rate_limited) is retried later.
+      parameters: [{ $ref: "#/components/parameters/ContributionID" }]
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              allOf:
+                - $ref: "#/components/schemas/Envelope"
+                - properties: { input: { $ref: "#/components/schemas/ParseInput" } }
+      responses:
+        "200":
+          description: The parsed document
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [output]
+                properties: { output: { $ref: "#/components/schemas/ParseOutput" } }
+        default: { $ref: "#/components/responses/Error" }
 components:
   securitySchemes:
     hostToken: { type: http, scheme: bearer }
@@ -296,6 +323,29 @@ components:
         metadata: { type: object, additionalProperties: { type: string } }
         isDeleted: { type: boolean }
         sourceResourceId: { type: string }
+    ParseInput:
+      type: object
+      properties:
+        fileName: { type: string }
+        fileType: { type: string, description: "Lower-case extension without the dot" }
+        content: { type: string, contentEncoding: base64 }
+        url: { type: string }
+        title: { type: string }
+    ParseOutput:
+      type: object
+      required: [markdown]
+      properties:
+        markdown: { type: string }
+        images:
+          type: array
+          items:
+            type: object
+            required: [originalRef, data]
+            properties:
+              originalRef: { type: string }
+              mimeType: { type: string }
+              data: { type: string, contentEncoding: base64 }
+        metadata: { type: object, additionalProperties: { type: string } }
     Event:
       type: object
       required: [type]
diff --git a/pluginsdk/pluginapi/parser.go b/pluginsdk/pluginapi/parser.go
new file mode 100644
index 000000000..9b9379e48
--- /dev/null
+++ b/pluginsdk/pluginapi/parser.go
@@ -0,0 +1,34 @@
+package pluginapi
+
+// ParsePath is the parse endpoint of parser id.
+func ParsePath(id string) string { return "/v1/parsers/" + id + "/parse" }
+
+// ParseInput is one document to parse: its bytes (file mode) or a URL (URL
+// mode). The parser's own settings arrive in Config.System and Config.Tenant.
+type ParseInput struct {
+	FileName string `json:"fileName,omitempty"`
+	// FileType is the lower-case extension without the dot ("pdf").
+	FileType string `json:"fileType,omitempty"`
+	// Content is the file; JSON carries it as base64.
+	Content []byte `json:"content,omitempty"`
+	URL     string `json:"url,omitempty"`
+	Title   string `json:"title,omitempty"`
+}
+
+// ParseOutput is the document as Markdown. WeKnora chunks it; parsers do not.
+type ParseOutput struct {
+	Markdown string `json:"markdown"`
+	// Images are the pictures the Markdown references: an image whose
+	// OriginalRef equals a ![](target) in Markdown is stored and the link
+	// rewritten. Inline data: URIs in the Markdown work too.
+	Images []ParsedImage `json:"images,omitempty"`
+	// Metadata such as "title" (URL mode) or "pages".
+	Metadata map[string]string `json:"metadata,omitempty"`
+}
+
+// ParsedImage is one image of a parsed document.
+type ParsedImage struct {
+	OriginalRef string `json:"originalRef"`
+	MimeType    string `json:"mimeType,omitempty"`
+	Data        []byte `json:"data"`
+}