Agent 记忆

1
2
3
4
5
6
7
8
9
10{
  "mcpServers": {
    "ov-mcp-server": {
      "url": "https://api.vikingdb.cn-beijing.volces.com/openviking/mcp",
      "headers": {
        "Authorization": "Bearer <OPENVIKING_API_KEY>"
      }
    }
  }
}

1ov health

1pip install openviking --upgrade --force-reinstall

1export OPENVIKING_API_KEY="your-api-key"

1
2
3
4
5
6
7
8
9
10import os
import openviking as ov

client = ov.SyncHTTPClient(
    url="https://api.vikingdb.cn-beijing.volces.com/openviking",
    api_key=os.environ["OPENVIKING_API_KEY"],
    agent_id="your-agent-id", # 自定义 agent_id，用于隔离 Agent 侧上下文
    timeout=120.0,
)
client.initialize()

1
2
3
4
5
6
7
8
9
10
11
12file_path = "your-file-path"
resource_to = "your-resource-path"  # e.g. viking://resources/demo/api.md
reason = "your-reason"  # e.g. External API documentation

result = client.add_resource(
    path=file_path,
    to=resource_to,
    reason=reason,
    wait=True,
)

print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13text = "your-message-text"  # e.g. I am a developer

session = client.create_session()
session_id = session["result"]["session_id"] if "result" in session else session["session_id"]

client.add_message(
    session_id,
    "user",
    parts=[{"type": "text", "text": text}],
)

result = client.commit_session(session_id)
print(result)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59import json
from pathlib import Path

import requests

url = "https://api.vikingdb.cn-beijing.volces.com/openviking"
api_key = "your-api-key" #可在用户管理页面复制
agent_id = "your-agent-id" #自定义agent-id
file_path = Path("your-file-path")  # e.g. test.txt
resource_to = "your-resource-path"  # e.g. viking://resources/demo/test.txt
reason = "your-reason"  # e.g. External API documentation

headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer " + api_key,
    "X-OpenViking-Agent": agent_id,
}


def post_json(path: str, payload: dict, timeout: float):
    response = requests.post(
        f"{url}{path}",
        headers=headers,
        json=payload,
        timeout=timeout,
    )
    response.raise_for_status()
    return response.json()


with file_path.open("rb") as file:
    result = requests.post(
        f"{url}/api/v1/resources/temp_upload",
        headers={
            "Authorization": "Bearer " + api_key,
            "X-OpenViking-Agent": agent_id,
        },
        files={"file": (file_path.name, file, "application/octet-stream")},
        timeout=120.0,
    )

result.raise_for_status()
result = result.json()
print(json.dumps(result, ensure_ascii=False, indent=2))

temp_file_id = result["result"]["temp_file_id"]

result = post_json(
    "/api/v1/resources",
    {
        "temp_file_id": temp_file_id,
        "source_name": file_path.name,
        "to": resource_to,
        "reason": reason,
    },
    120.0,
)

print(json.dumps(result, ensure_ascii=False, indent=2))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45import json
import requests

url = "https://api.vikingdb.cn-beijing.volces.com/openviking"
api_key = "your-api-key" #可在【用户管理页面】复制
agent_id = "your-agent-id" #自定义agent-id
text = "your-message-text"  # e.g. I am a developer

headers = {
    "Content-Type": "application/json",
    "Authorization": "Bearer " + api_key,
    "X-OpenViking-Agent": agent_id,
}


def post_json(path: str, payload: dict, timeout: float):
    response = requests.post(
        f"{url}{path}",
        headers=headers,
        json=payload,
        timeout=timeout,
    )
    response.raise_for_status()
    return response.json()


session = post_json("/api/v1/sessions", {}, 360.0)
session_id = session["result"]["session_id"]

post_json(
    f"/api/v1/sessions/{session_id}/messages",
    {
        "role": "user",
        "parts": [{"type": "text", "text": text}],
    },
    360.0,
)

result = post_json(
    f"/api/v1/sessions/{session_id}/commit",
    {"telemetry": False},
    360.0,
)

print(json.dumps(result, ensure_ascii=False, indent=2))