Agentic LLM

Build applications that autonomously call agents from Agentverse marketplace for complex workflows.

Build applications with ASI:One’s agentic models that can autonomously call agents from the Agentverse marketplace and handle complex workflows. These models can discover, coordinate, and execute tasks through a network of specialized agents available on Agentverse.

Overview

ASI:One’s agentic model (asi1) is designed to automatically discover and coordinate with agents from the Agentverse marketplace to accomplish complex tasks. It handles agent selection, orchestration, and execution planning autonomously by connecting to the vast ecosystem of agents available on Agentverse.

Key Features:

Autonomous Agent Discovery: Automatically finds relevant agents from Agentverse marketplace for your tasks
Session Persistence: Maintains conversation context across multiple interactions
Asynchronous Processing: Handles long-running agent workflows from Agentverse
Streaming Support: Real-time response streaming for better UX

Quick Start

Python

cURL

JavaScript

1 import os
2 import uuid
3 import json
4 import requests
5 import sys
6 import time
7 
8 API_KEY = os.getenv("ASI_ONE_API_KEY") or "sk-REPLACE_ME"
9 ENDPOINT = "https://api.asi1.ai/v1/chat/completions"
10 MODEL = "asi1"
11 TIMEOUT = 90 # seconds
12 
13 # In-memory session management
14 
15 SESSION_MAP: dict[str, str] = {}
16 
17 def get_session_id(conv_id: str) -> str:
18     """Return existing session UUID for this conversation or create a new one."""
19     sid = SESSION_MAP.get(conv_id)
20     if sid is None:
21         sid = str(uuid.uuid4())
22         SESSION_MAP[conv_id] = sid
23     return sid
24 
25 def ask(conv_id: str, messages: list[dict], *, stream: bool = False) -> str:
26     """Send the messages list to the ASI:One agent and return the assistant reply."""
27     session_id = get_session_id(conv_id)
28     print(f"[session] Using session-id: {session_id}")
29 
30     headers = {
31         "Authorization": f"Bearer {API_KEY}",
32         "x-session-id": session_id,
33         "Content-Type": "application/json",
34     }
35 
36     payload = {
37         "model": MODEL,
38         "messages": messages,
39         "stream": stream,
40     }
41 
42     if not stream:
43         resp = requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT)
44         resp.raise_for_status()
45         return resp.json()["choices"][0]["message"]["content"]
46 
47     # Streaming implementation
48     with requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT, stream=True) as resp:
49         resp.raise_for_status()
50         full_text = ""
51         for line in resp.iter_lines(decode_unicode=True):
52             if not line or not line.startswith("data: "):
53                 continue
54             line = line[len("data: ") :]
55             if line == "[DONE]":
56                 break
57             try:
58                 chunk = json.loads(line)
59                 choices = chunk.get("choices")
60                 if choices and "content" in choices[0].get("delta", {}):
61                     token = choices[0]["delta"]["content"]
62                     sys.stdout.write(token)
63                     sys.stdout.flush()
64                     full_text += token
65             except json.JSONDecodeError:
66                 continue
67         print()
68         return full_text
69 
70 if __name__ == "__main__": # Simple usage example
71     conv_id = str(uuid.uuid4())
72     messages = [
73         {"role": "user", "content": "use Hi-dream model to generate image of monkey sitting on top of mountain"}
74     ]
75     reply = ask(conv_id, messages, stream=True)
76     print(f"\nAssistant: {reply}")

Example Output

[session] Using session-id: d92b1ff5-3be0-484d-afe4-04edc5239a1c
I'll generate an image of a monkey sitting on top of a mountain for you using the Hi-dream model.
Image generated. ![generated-image](https://res.cloudinary.com/fetch-ai/image/upload/v1768411116/49557c0f2a3a4b258f1f070b55bdf921.png.jpg)

The exact wording and session-ID will vary, but you should always receive a direct image link once generation completes.

Session Management

Agentic models require session persistence to maintain context across agent interactions with the Agentverse marketplace. Always include the x-session-id header:

1 import uuid
2 
3 # Create or retrieve session ID for conversation
4 def get_session_id(conversation_id: str) -> str:
5     # In production, store this in Redis or database
6     session_id = SESSION_MAP.get(conversation_id)
7     if not session_id:
8         session_id = str(uuid.uuid4())
9         SESSION_MAP[conversation_id] = session_id
10     return session_id
11 
12 # Include in every request
13 headers = {
14     "Authorization": f"Bearer {API_KEY}",
15     "x-session-id": get_session_id("user_123_chat"),
16     "Content-Type": "application/json"
17 }

Asynchronous Agent Processing

When agents from Agentverse marketplace need time to complete tasks, the model may send a deferred response. Poll for updates:

1 import uuid
2 import time
3 
4 # Create or retrieve session ID for conversation
5 def get_session_id(conversation_id: str) -> str:
6     # In production, store this in Redis or database
7     session_id = SESSION_MAP.get(conversation_id)
8     if not session_id:
9         session_id = str(uuid.uuid4())
10         SESSION_MAP[conversation_id] = session_id
11     return session_id
12 
13 # Include in every request
14 headers = {
15     "Authorization": f"Bearer {API_KEY}",
16     "x-session-id": get_session_id("user_123_chat"),
17     "Content-Type": "application/json"
18 }
19 
20 def poll_for_async_reply(
21     conv_id: str,
22     history: list[dict],
23     *,
24     wait_sec: int = 5,  # poll every 5 seconds
25     max_attempts: int = 24,  # ~2 minutes total
26 ) -> str | None:
27     """Ask ASI:One 'Any update?' until reply text actually changes."""
28     for attempt in range(max_attempts):
29         time.sleep(wait_sec)
30         print(f"🔄 polling (attempt {attempt + 1}) …", flush=True)
31         update_prompt = {"role": "user", "content": "Any update?"}
32         latest = ask(conv_id, history + [update_prompt], stream=False)
33         if latest and latest.strip() != history[-1]["content"].strip():
34             return latest
35     return None
36 
37 # Usage example after receiving the initial deferred reply
38 assistant_reply = ask(conv_id, messages, stream=False)
39 history = messages + [{"role": "assistant", "content": assistant_reply}]
40 
41 if assistant_reply.strip() == "I've sent the message":
42     follow_up = poll_for_async_reply(conv_id, history)
43     if follow_up:
44         print(f"Agentverse agent completed task: {follow_up}")
45         history.append({"role": "assistant", "content": follow_up})

Best Practices

Session Management

Use UUIDs for session IDs to avoid collisions
Store session mappings in Redis or database for production
Include x-session-id header in every request to maintain context

Error Handling

Implement timeouts for long-running agent tasks
Handle network failures with exponential backoff
Validate responses before processing agent results

Performance Optimization

Use streaming for better user experience
Implement async polling for deferred agent responses

Agent Coordination

Be specific in requests to help agent discovery from Agentverse marketplace
Allow time for complex multi-agent workflows involving Agentverse agents
Monitor session state to understand Agentverse agent progress

Build applications that autonomously call agents from Agentverse marketplace for complex workflows.

Overview

Key Features:

Autonomous Agent Discovery: Automatically finds relevant agents from Agentverse marketplace for your tasks
Session Persistence: Maintains conversation context across multiple interactions
Asynchronous Processing: Handles long-running agent workflows from Agentverse
Streaming Support: Real-time response streaming for better UX

Quick Start

Python

cURL

JavaScript

1 import os
2 import uuid
3 import json
4 import requests
5 import sys
6 import time
7 
8 API_KEY = os.getenv("ASI_ONE_API_KEY") or "sk-REPLACE_ME"
9 ENDPOINT = "https://api.asi1.ai/v1/chat/completions"
10 MODEL = "asi1"
11 TIMEOUT = 90 # seconds
12 
13 # In-memory session management
14 
15 SESSION_MAP: dict[str, str] = {}
16 
17 def get_session_id(conv_id: str) -> str:
18     """Return existing session UUID for this conversation or create a new one."""
19     sid = SESSION_MAP.get(conv_id)
20     if sid is None:
21         sid = str(uuid.uuid4())
22         SESSION_MAP[conv_id] = sid
23     return sid
24 
25 def ask(conv_id: str, messages: list[dict], *, stream: bool = False) -> str:
26     """Send the messages list to the ASI:One agent and return the assistant reply."""
27     session_id = get_session_id(conv_id)
28     print(f"[session] Using session-id: {session_id}")
29 
30     headers = {
31         "Authorization": f"Bearer {API_KEY}",
32         "x-session-id": session_id,
33         "Content-Type": "application/json",
34     }
35 
36     payload = {
37         "model": MODEL,
38         "messages": messages,
39         "stream": stream,
40     }
41 
42     if not stream:
43         resp = requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT)
44         resp.raise_for_status()
45         return resp.json()["choices"][0]["message"]["content"]
46 
47     # Streaming implementation
48     with requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT, stream=True) as resp:
49         resp.raise_for_status()
50         full_text = ""
51         for line in resp.iter_lines(decode_unicode=True):
52             if not line or not line.startswith("data: "):
53                 continue
54             line = line[len("data: ") :]
55             if line == "[DONE]":
56                 break
57             try:
58                 chunk = json.loads(line)
59                 choices = chunk.get("choices")
60                 if choices and "content" in choices[0].get("delta", {}):
61                     token = choices[0]["delta"]["content"]
62                     sys.stdout.write(token)
63                     sys.stdout.flush()
64                     full_text += token
65             except json.JSONDecodeError:
66                 continue
67         print()
68         return full_text
69 
70 if __name__ == "__main__": # Simple usage example
71     conv_id = str(uuid.uuid4())
72     messages = [
73         {"role": "user", "content": "use Hi-dream model to generate image of monkey sitting on top of mountain"}
74     ]
75     reply = ask(conv_id, messages, stream=True)
76     print(f"\nAssistant: {reply}")

Example Output

[session] Using session-id: d92b1ff5-3be0-484d-afe4-04edc5239a1c
I'll generate an image of a monkey sitting on top of a mountain for you using the Hi-dream model.
Image generated. ![generated-image](https://res.cloudinary.com/fetch-ai/image/upload/v1768411116/49557c0f2a3a4b258f1f070b55bdf921.png.jpg)

The exact wording and session-ID will vary, but you should always receive a direct image link once generation completes.

Session Management

Agentic models require session persistence to maintain context across agent interactions with the Agentverse marketplace. Always include the x-session-id header:

1 import uuid
2 
3 # Create or retrieve session ID for conversation
4 def get_session_id(conversation_id: str) -> str:
5     # In production, store this in Redis or database
6     session_id = SESSION_MAP.get(conversation_id)
7     if not session_id:
8         session_id = str(uuid.uuid4())
9         SESSION_MAP[conversation_id] = session_id
10     return session_id
11 
12 # Include in every request
13 headers = {
14     "Authorization": f"Bearer {API_KEY}",
15     "x-session-id": get_session_id("user_123_chat"),
16     "Content-Type": "application/json"
17 }

Asynchronous Agent Processing

When agents from Agentverse marketplace need time to complete tasks, the model may send a deferred response. Poll for updates:

1 import uuid
2 import time
3 
4 # Create or retrieve session ID for conversation
5 def get_session_id(conversation_id: str) -> str:
6     # In production, store this in Redis or database
7     session_id = SESSION_MAP.get(conversation_id)
8     if not session_id:
9         session_id = str(uuid.uuid4())
10         SESSION_MAP[conversation_id] = session_id
11     return session_id
12 
13 # Include in every request
14 headers = {
15     "Authorization": f"Bearer {API_KEY}",
16     "x-session-id": get_session_id("user_123_chat"),
17     "Content-Type": "application/json"
18 }
19 
20 def poll_for_async_reply(
21     conv_id: str,
22     history: list[dict],
23     *,
24     wait_sec: int = 5,  # poll every 5 seconds
25     max_attempts: int = 24,  # ~2 minutes total
26 ) -> str | None:
27     """Ask ASI:One 'Any update?' until reply text actually changes."""
28     for attempt in range(max_attempts):
29         time.sleep(wait_sec)
30         print(f"🔄 polling (attempt {attempt + 1}) …", flush=True)
31         update_prompt = {"role": "user", "content": "Any update?"}
32         latest = ask(conv_id, history + [update_prompt], stream=False)
33         if latest and latest.strip() != history[-1]["content"].strip():
34             return latest
35     return None
36 
37 # Usage example after receiving the initial deferred reply
38 assistant_reply = ask(conv_id, messages, stream=False)
39 history = messages + [{"role": "assistant", "content": assistant_reply}]
40 
41 if assistant_reply.strip() == "I've sent the message":
42     follow_up = poll_for_async_reply(conv_id, history)
43     if follow_up:
44         print(f"Agentverse agent completed task: {follow_up}")
45         history.append({"role": "assistant", "content": follow_up})

Best Practices

Session Management

Use UUIDs for session IDs to avoid collisions
Store session mappings in Redis or database for production
Include x-session-id header in every request to maintain context

Error Handling

Implement timeouts for long-running agent tasks
Handle network failures with exponential backoff
Validate responses before processing agent results

Performance Optimization

Use streaming for better user experience
Implement async polling for deferred agent responses

Agent Coordination

Be specific in requests to help agent discovery from Agentverse marketplace
Allow time for complex multi-agent workflows involving Agentverse agents
Monitor session state to understand Agentverse agent progress

1	import os
2	import uuid
3	import json
4	import requests
5	import sys
6	import time
7
8	API_KEY = os.getenv("ASI_ONE_API_KEY") or "sk-REPLACE_ME"
9	ENDPOINT = "https://api.asi1.ai/v1/chat/completions"
10	MODEL = "asi1"
11	TIMEOUT = 90 # seconds
12
13	# In-memory session management
14
15	SESSION_MAP: dict[str, str] = {}
16
17	def get_session_id(conv_id: str) -> str:
18	"""Return existing session UUID for this conversation or create a new one."""
19	sid = SESSION_MAP.get(conv_id)
20	if sid is None:
21	sid = str(uuid.uuid4())
22	SESSION_MAP[conv_id] = sid
23	return sid
24
25	def ask(conv_id: str, messages: list[dict], *, stream: bool = False) -> str:
26	"""Send the messages list to the ASI:One agent and return the assistant reply."""
27	session_id = get_session_id(conv_id)
28	print(f"[session] Using session-id: {session_id}")
29
30	headers = {
31	"Authorization": f"Bearer {API_KEY}",
32	"x-session-id": session_id,
33	"Content-Type": "application/json",
34	}
35
36	payload = {
37	"model": MODEL,
38	"messages": messages,
39	"stream": stream,
40	}
41
42	if not stream:
43	resp = requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT)
44	resp.raise_for_status()
45	return resp.json()["choices"][0]["message"]["content"]
46
47	# Streaming implementation
48	with requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT, stream=True) as resp:
49	resp.raise_for_status()
50	full_text = ""
51	for line in resp.iter_lines(decode_unicode=True):
52	if not line or not line.startswith("data: "):
53	continue
54	line = line[len("data: ") :]
55	if line == "[DONE]":
56	break
57	try:
58	chunk = json.loads(line)
59	choices = chunk.get("choices")
60	if choices and "content" in choices[0].get("delta", {}):
61	token = choices[0]["delta"]["content"]
62	sys.stdout.write(token)
63	sys.stdout.flush()
64	full_text += token
65	except json.JSONDecodeError:
66	continue
67	print()
68	return full_text
69
70	if __name__ == "__main__": # Simple usage example
71	conv_id = str(uuid.uuid4())
72	messages = [
73	{"role": "user", "content": "use Hi-dream model to generate image of monkey sitting on top of mountain"}
74	]
75	reply = ask(conv_id, messages, stream=True)
76	print(f"\nAssistant: {reply}")

1	import uuid
2
3	# Create or retrieve session ID for conversation
4	def get_session_id(conversation_id: str) -> str:
5	# In production, store this in Redis or database
6	session_id = SESSION_MAP.get(conversation_id)
7	if not session_id:
8	session_id = str(uuid.uuid4())
9	SESSION_MAP[conversation_id] = session_id
10	return session_id
11
12	# Include in every request
13	headers = {
14	"Authorization": f"Bearer {API_KEY}",
15	"x-session-id": get_session_id("user_123_chat"),
16	"Content-Type": "application/json"
17	}